1
# Copyright (C) 2005, 2006, 2007, 2008 Canonical Ltd
3
# This program is free software; you can redistribute it and/or modify
4
# it under the terms of the GNU General Public License as published by
5
# the Free Software Foundation; either version 2 of the License, or
6
# (at your option) any later version.
8
# This program is distributed in the hope that it will be useful,
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
# GNU General Public License for more details.
13
# You should have received a copy of the GNU General Public License
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
# TODO: Perhaps there should be an API to find out if bzr running under the
19
# test suite -- some plugins might want to avoid making intrusive changes if
20
# this is the case. However, we want behaviour under to test to diverge as
21
# little as possible, so this should be used rarely if it's added at all.
22
# (Suggestion from j-a-meinel, 2005-11-24)
24
# NOTE: Some classes in here use camelCaseNaming() rather than
25
# underscore_naming(). That's for consistency with unittest; it's not the
26
# general style of bzrlib. Please continue that consistency when adding e.g.
27
# new assertFoo() methods.
31
from cStringIO import StringIO
37
from pprint import pformat
42
from subprocess import Popen, PIPE
63
import bzrlib.commands
64
import bzrlib.timestamp
66
import bzrlib.inventory
67
import bzrlib.iterablefile
72
# lsprof not available
74
from bzrlib.merge import merge_inner
78
from bzrlib import symbol_versioning
79
from bzrlib.symbol_versioning import (
86
from bzrlib.transport import get_transport
87
import bzrlib.transport
88
from bzrlib.transport.local import LocalURLServer
89
from bzrlib.transport.memory import MemoryServer
90
from bzrlib.transport.readonly import ReadonlyServer
91
from bzrlib.trace import mutter, note
92
from bzrlib.tests import TestUtil
93
from bzrlib.tests.http_server import HttpServer
94
from bzrlib.tests.TestUtil import (
98
from bzrlib.tests.treeshape import build_tree_contents
99
import bzrlib.version_info_formats.format_custom
100
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
102
# Mark this python module as being part of the implementation
103
# of unittest: this gives us better tracebacks where the last
104
# shown frame is the test code, not our assertXYZ.
107
default_transport = LocalURLServer
110
class ExtendedTestResult(unittest._TextTestResult):
111
"""Accepts, reports and accumulates the results of running tests.
113
Compared to the unittest version this class adds support for
114
profiling, benchmarking, stopping as soon as a test fails, and
115
skipping tests. There are further-specialized subclasses for
116
different types of display.
118
When a test finishes, in whatever way, it calls one of the addSuccess,
119
addFailure or addError classes. These in turn may redirect to a more
120
specific case for the special test results supported by our extended
123
Note that just one of these objects is fed the results from many tests.
128
def __init__(self, stream, descriptions, verbosity,
132
"""Construct new TestResult.
134
:param bench_history: Optionally, a writable file object to accumulate
137
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
138
if bench_history is not None:
139
from bzrlib.version import _get_bzr_source_tree
140
src_tree = _get_bzr_source_tree()
143
revision_id = src_tree.get_parent_ids()[0]
145
# XXX: if this is a brand new tree, do the same as if there
149
# XXX: If there's no branch, what should we do?
151
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
152
self._bench_history = bench_history
153
self.ui = ui.ui_factory
154
self.num_tests = num_tests
156
self.failure_count = 0
157
self.known_failure_count = 0
159
self.not_applicable_count = 0
160
self.unsupported = {}
162
self._overall_start_time = time.time()
164
def _extractBenchmarkTime(self, testCase):
165
"""Add a benchmark time for the current test case."""
166
return getattr(testCase, "_benchtime", None)
168
def _elapsedTestTimeString(self):
169
"""Return a time string for the overall time the current test has taken."""
170
return self._formatTime(time.time() - self._start_time)
172
def _testTimeString(self, testCase):
173
benchmark_time = self._extractBenchmarkTime(testCase)
174
if benchmark_time is not None:
176
self._formatTime(benchmark_time),
177
self._elapsedTestTimeString())
179
return " %s" % self._elapsedTestTimeString()
181
def _formatTime(self, seconds):
182
"""Format seconds as milliseconds with leading spaces."""
183
# some benchmarks can take thousands of seconds to run, so we need 8
185
return "%8dms" % (1000 * seconds)
187
def _shortened_test_description(self, test):
189
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
192
def startTest(self, test):
193
unittest.TestResult.startTest(self, test)
194
self.report_test_start(test)
195
test.number = self.count
196
self._recordTestStartTime()
198
def _recordTestStartTime(self):
199
"""Record that a test has started."""
200
self._start_time = time.time()
202
def _cleanupLogFile(self, test):
203
# We can only do this if we have one of our TestCases, not if
205
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
206
if setKeepLogfile is not None:
209
def addError(self, test, err):
210
"""Tell result that test finished with an error.
212
Called from the TestCase run() method when the test
213
fails with an unexpected error.
215
self._testConcluded(test)
216
if isinstance(err[1], TestSkipped):
217
return self._addSkipped(test, err)
218
elif isinstance(err[1], UnavailableFeature):
219
return self.addNotSupported(test, err[1].args[0])
221
self._cleanupLogFile(test)
222
unittest.TestResult.addError(self, test, err)
223
self.error_count += 1
224
self.report_error(test, err)
228
def addFailure(self, test, err):
229
"""Tell result that test failed.
231
Called from the TestCase run() method when the test
232
fails because e.g. an assert() method failed.
234
self._testConcluded(test)
235
if isinstance(err[1], KnownFailure):
236
return self._addKnownFailure(test, err)
238
self._cleanupLogFile(test)
239
unittest.TestResult.addFailure(self, test, err)
240
self.failure_count += 1
241
self.report_failure(test, err)
245
def addSuccess(self, test):
246
"""Tell result that test completed successfully.
248
Called from the TestCase run()
250
self._testConcluded(test)
251
if self._bench_history is not None:
252
benchmark_time = self._extractBenchmarkTime(test)
253
if benchmark_time is not None:
254
self._bench_history.write("%s %s\n" % (
255
self._formatTime(benchmark_time),
257
self.report_success(test)
258
self._cleanupLogFile(test)
259
unittest.TestResult.addSuccess(self, test)
260
test._log_contents = ''
262
def _testConcluded(self, test):
263
"""Common code when a test has finished.
265
Called regardless of whether it succeded, failed, etc.
269
def _addKnownFailure(self, test, err):
270
self.known_failure_count += 1
271
self.report_known_failure(test, err)
273
def addNotSupported(self, test, feature):
274
"""The test will not be run because of a missing feature.
276
# this can be called in two different ways: it may be that the
277
# test started running, and then raised (through addError)
278
# UnavailableFeature. Alternatively this method can be called
279
# while probing for features before running the tests; in that
280
# case we will see startTest and stopTest, but the test will never
282
self.unsupported.setdefault(str(feature), 0)
283
self.unsupported[str(feature)] += 1
284
self.report_unsupported(test, feature)
286
def _addSkipped(self, test, skip_excinfo):
287
if isinstance(skip_excinfo[1], TestNotApplicable):
288
self.not_applicable_count += 1
289
self.report_not_applicable(test, skip_excinfo)
292
self.report_skip(test, skip_excinfo)
295
except KeyboardInterrupt:
298
self.addError(test, test._exc_info())
300
# seems best to treat this as success from point-of-view of unittest
301
# -- it actually does nothing so it barely matters :)
302
unittest.TestResult.addSuccess(self, test)
303
test._log_contents = ''
305
def printErrorList(self, flavour, errors):
306
for test, err in errors:
307
self.stream.writeln(self.separator1)
308
self.stream.write("%s: " % flavour)
309
self.stream.writeln(self.getDescription(test))
310
if getattr(test, '_get_log', None) is not None:
311
self.stream.write('\n')
313
('vvvv[log from %s]' % test.id()).ljust(78,'-'))
314
self.stream.write('\n')
315
self.stream.write(test._get_log())
316
self.stream.write('\n')
318
('^^^^[log from %s]' % test.id()).ljust(78,'-'))
319
self.stream.write('\n')
320
self.stream.writeln(self.separator2)
321
self.stream.writeln("%s" % err)
326
def report_cleaning_up(self):
329
def report_success(self, test):
332
def wasStrictlySuccessful(self):
333
if self.unsupported or self.known_failure_count:
335
return self.wasSuccessful()
338
class TextTestResult(ExtendedTestResult):
339
"""Displays progress and results of tests in text form"""
341
def __init__(self, stream, descriptions, verbosity,
346
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
347
bench_history, num_tests)
349
self.pb = self.ui.nested_progress_bar()
350
self._supplied_pb = False
353
self._supplied_pb = True
354
self.pb.show_pct = False
355
self.pb.show_spinner = False
356
self.pb.show_eta = False,
357
self.pb.show_count = False
358
self.pb.show_bar = False
360
def report_starting(self):
361
self.pb.update('[test 0/%d] starting...' % (self.num_tests))
363
def _progress_prefix_text(self):
364
# the longer this text, the less space we have to show the test
366
a = '[%d' % self.count # total that have been run
367
# tests skipped as known not to be relevant are not important enough
369
## if self.skip_count:
370
## a += ', %d skip' % self.skip_count
371
## if self.known_failure_count:
372
## a += '+%dX' % self.known_failure_count
373
if self.num_tests is not None:
374
a +='/%d' % self.num_tests
376
runtime = time.time() - self._overall_start_time
378
a += '%dm%ds' % (runtime / 60, runtime % 60)
382
a += ', %d err' % self.error_count
383
if self.failure_count:
384
a += ', %d fail' % self.failure_count
386
a += ', %d missing' % len(self.unsupported)
390
def report_test_start(self, test):
393
self._progress_prefix_text()
395
+ self._shortened_test_description(test))
397
def _test_description(self, test):
398
return self._shortened_test_description(test)
400
def report_error(self, test, err):
401
self.pb.note('ERROR: %s\n %s\n',
402
self._test_description(test),
406
def report_failure(self, test, err):
407
self.pb.note('FAIL: %s\n %s\n',
408
self._test_description(test),
412
def report_known_failure(self, test, err):
413
self.pb.note('XFAIL: %s\n%s\n',
414
self._test_description(test), err[1])
416
def report_skip(self, test, skip_excinfo):
419
def report_not_applicable(self, test, skip_excinfo):
422
def report_unsupported(self, test, feature):
423
"""test cannot be run because feature is missing."""
425
def report_cleaning_up(self):
426
self.pb.update('cleaning up...')
429
if not self._supplied_pb:
433
class VerboseTestResult(ExtendedTestResult):
434
"""Produce long output, with one line per test run plus times"""
436
def _ellipsize_to_right(self, a_string, final_width):
437
"""Truncate and pad a string, keeping the right hand side"""
438
if len(a_string) > final_width:
439
result = '...' + a_string[3-final_width:]
442
return result.ljust(final_width)
444
def report_starting(self):
445
self.stream.write('running %d tests...\n' % self.num_tests)
447
def report_test_start(self, test):
449
name = self._shortened_test_description(test)
450
# width needs space for 6 char status, plus 1 for slash, plus 2 10-char
451
# numbers, plus a trailing blank
452
# when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on space
453
self.stream.write(self._ellipsize_to_right(name,
454
osutils.terminal_width()-30))
457
def _error_summary(self, err):
459
return '%s%s' % (indent, err[1])
461
def report_error(self, test, err):
462
self.stream.writeln('ERROR %s\n%s'
463
% (self._testTimeString(test),
464
self._error_summary(err)))
466
def report_failure(self, test, err):
467
self.stream.writeln(' FAIL %s\n%s'
468
% (self._testTimeString(test),
469
self._error_summary(err)))
471
def report_known_failure(self, test, err):
472
self.stream.writeln('XFAIL %s\n%s'
473
% (self._testTimeString(test),
474
self._error_summary(err)))
476
def report_success(self, test):
477
self.stream.writeln(' OK %s' % self._testTimeString(test))
478
for bench_called, stats in getattr(test, '_benchcalls', []):
479
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
480
stats.pprint(file=self.stream)
481
# flush the stream so that we get smooth output. This verbose mode is
482
# used to show the output in PQM.
485
def report_skip(self, test, skip_excinfo):
486
self.stream.writeln(' SKIP %s\n%s'
487
% (self._testTimeString(test),
488
self._error_summary(skip_excinfo)))
490
def report_not_applicable(self, test, skip_excinfo):
491
self.stream.writeln(' N/A %s\n%s'
492
% (self._testTimeString(test),
493
self._error_summary(skip_excinfo)))
495
def report_unsupported(self, test, feature):
496
"""test cannot be run because feature is missing."""
497
self.stream.writeln("NODEP %s\n The feature '%s' is not available."
498
%(self._testTimeString(test), feature))
501
class TextTestRunner(object):
502
stop_on_failure = False
511
self.stream = unittest._WritelnDecorator(stream)
512
self.descriptions = descriptions
513
self.verbosity = verbosity
514
self._bench_history = bench_history
515
self.list_only = list_only
518
"Run the given test case or test suite."
519
startTime = time.time()
520
if self.verbosity == 1:
521
result_class = TextTestResult
522
elif self.verbosity >= 2:
523
result_class = VerboseTestResult
524
result = result_class(self.stream,
527
bench_history=self._bench_history,
528
num_tests=test.countTestCases(),
530
result.stop_early = self.stop_on_failure
531
result.report_starting()
533
if self.verbosity >= 2:
534
self.stream.writeln("Listing tests only ...\n")
536
for t in iter_suite_tests(test):
537
self.stream.writeln("%s" % (t.id()))
539
actionTaken = "Listed"
542
run = result.testsRun
544
stopTime = time.time()
545
timeTaken = stopTime - startTime
547
self.stream.writeln(result.separator2)
548
self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
549
run, run != 1 and "s" or "", timeTaken))
550
self.stream.writeln()
551
if not result.wasSuccessful():
552
self.stream.write("FAILED (")
553
failed, errored = map(len, (result.failures, result.errors))
555
self.stream.write("failures=%d" % failed)
557
if failed: self.stream.write(", ")
558
self.stream.write("errors=%d" % errored)
559
if result.known_failure_count:
560
if failed or errored: self.stream.write(", ")
561
self.stream.write("known_failure_count=%d" %
562
result.known_failure_count)
563
self.stream.writeln(")")
565
if result.known_failure_count:
566
self.stream.writeln("OK (known_failures=%d)" %
567
result.known_failure_count)
569
self.stream.writeln("OK")
570
if result.skip_count > 0:
571
skipped = result.skip_count
572
self.stream.writeln('%d test%s skipped' %
573
(skipped, skipped != 1 and "s" or ""))
574
if result.unsupported:
575
for feature, count in sorted(result.unsupported.items()):
576
self.stream.writeln("Missing feature '%s' skipped %d tests." %
582
def iter_suite_tests(suite):
583
"""Return all tests in a suite, recursing through nested suites"""
584
for item in suite._tests:
585
if isinstance(item, unittest.TestCase):
587
elif isinstance(item, unittest.TestSuite):
588
for r in iter_suite_tests(item):
591
raise Exception('unknown object %r inside test suite %r'
595
class TestSkipped(Exception):
596
"""Indicates that a test was intentionally skipped, rather than failing."""
599
class TestNotApplicable(TestSkipped):
600
"""A test is not applicable to the situation where it was run.
602
This is only normally raised by parameterized tests, if they find that
603
the instance they're constructed upon does not support one aspect
608
class KnownFailure(AssertionError):
609
"""Indicates that a test failed in a precisely expected manner.
611
Such failures dont block the whole test suite from passing because they are
612
indicators of partially completed code or of future work. We have an
613
explicit error for them so that we can ensure that they are always visible:
614
KnownFailures are always shown in the output of bzr selftest.
618
class UnavailableFeature(Exception):
619
"""A feature required for this test was not available.
621
The feature should be used to construct the exception.
625
class CommandFailed(Exception):
629
class StringIOWrapper(object):
630
"""A wrapper around cStringIO which just adds an encoding attribute.
632
Internally we can check sys.stdout to see what the output encoding
633
should be. However, cStringIO has no encoding attribute that we can
634
set. So we wrap it instead.
639
def __init__(self, s=None):
641
self.__dict__['_cstring'] = StringIO(s)
643
self.__dict__['_cstring'] = StringIO()
645
def __getattr__(self, name, getattr=getattr):
646
return getattr(self.__dict__['_cstring'], name)
648
def __setattr__(self, name, val):
649
if name == 'encoding':
650
self.__dict__['encoding'] = val
652
return setattr(self._cstring, name, val)
655
class TestUIFactory(ui.CLIUIFactory):
656
"""A UI Factory for testing.
658
Hide the progress bar but emit note()s.
660
Allows get_password to be tested without real tty attached.
667
super(TestUIFactory, self).__init__()
668
if stdin is not None:
669
# We use a StringIOWrapper to be able to test various
670
# encodings, but the user is still responsible to
671
# encode the string and to set the encoding attribute
672
# of StringIOWrapper.
673
self.stdin = StringIOWrapper(stdin)
675
self.stdout = sys.stdout
679
self.stderr = sys.stderr
684
"""See progress.ProgressBar.clear()."""
686
def clear_term(self):
687
"""See progress.ProgressBar.clear_term()."""
689
def clear_term(self):
690
"""See progress.ProgressBar.clear_term()."""
693
"""See progress.ProgressBar.finished()."""
695
def note(self, fmt_string, *args, **kwargs):
696
"""See progress.ProgressBar.note()."""
697
self.stdout.write((fmt_string + "\n") % args)
699
def progress_bar(self):
702
def nested_progress_bar(self):
705
def update(self, message, count=None, total=None):
706
"""See progress.ProgressBar.update()."""
708
def get_non_echoed_password(self, prompt):
709
"""Get password from stdin without trying to handle the echo mode"""
711
self.stdout.write(prompt.encode(self.stdout.encoding, 'replace'))
712
password = self.stdin.readline()
715
if password[-1] == '\n':
716
password = password[:-1]
720
def _report_leaked_threads():
721
bzrlib.trace.warning('%s is leaking threads among %d leaking tests',
722
TestCase._first_thread_leaker_id,
723
TestCase._leaking_threads_tests)
726
class TestCase(unittest.TestCase):
727
"""Base class for bzr unit tests.
729
Tests that need access to disk resources should subclass
730
TestCaseInTempDir not TestCase.
732
Error and debug log messages are redirected from their usual
733
location into a temporary file, the contents of which can be
734
retrieved by _get_log(). We use a real OS file, not an in-memory object,
735
so that it can also capture file IO. When the test completes this file
736
is read into memory and removed from disk.
738
There are also convenience functions to invoke bzr's command-line
739
routine, and to build and check bzr trees.
741
In addition to the usual method of overriding tearDown(), this class also
742
allows subclasses to register functions into the _cleanups list, which is
743
run in order as the object is torn down. It's less likely this will be
744
accidentally overlooked.
747
_active_threads = None
748
_leaking_threads_tests = 0
749
_first_thread_leaker_id = None
750
_log_file_name = None
752
_keep_log_file = False
753
# record lsprof data when performing benchmark calls.
754
_gather_lsprof_in_benchmarks = False
755
attrs_to_keep = ('id', '_testMethodName', '_testMethodDoc',
756
'_log_contents', '_log_file_name', '_benchtime',
757
'_TestCase__testMethodName')
759
def __init__(self, methodName='testMethod'):
760
super(TestCase, self).__init__(methodName)
764
unittest.TestCase.setUp(self)
765
self._cleanEnvironment()
768
self._benchcalls = []
769
self._benchtime = None
771
self._clear_debug_flags()
772
TestCase._active_threads = threading.activeCount()
773
self.addCleanup(self._check_leaked_threads)
775
def _check_leaked_threads(self):
776
active = threading.activeCount()
777
leaked_threads = active - TestCase._active_threads
778
TestCase._active_threads = active
780
TestCase._leaking_threads_tests += 1
781
if TestCase._first_thread_leaker_id is None:
782
TestCase._first_thread_leaker_id = self.id()
783
# we're not specifically told when all tests are finished.
784
# This will do. We use a function to avoid keeping a reference
785
# to a TestCase object.
786
atexit.register(_report_leaked_threads)
788
def _clear_debug_flags(self):
789
"""Prevent externally set debug flags affecting tests.
791
Tests that want to use debug flags can just set them in the
792
debug_flags set during setup/teardown.
794
if 'allow_debug' not in selftest_debug_flags:
795
self._preserved_debug_flags = set(debug.debug_flags)
796
debug.debug_flags.clear()
797
self.addCleanup(self._restore_debug_flags)
799
def _clear_hooks(self):
800
# prevent hooks affecting tests
802
import bzrlib.smart.server
803
self._preserved_hooks = {
804
bzrlib.branch.Branch: bzrlib.branch.Branch.hooks,
805
bzrlib.mutabletree.MutableTree: bzrlib.mutabletree.MutableTree.hooks,
806
bzrlib.smart.server.SmartTCPServer: bzrlib.smart.server.SmartTCPServer.hooks,
808
self.addCleanup(self._restoreHooks)
809
# reset all hooks to an empty instance of the appropriate type
810
bzrlib.branch.Branch.hooks = bzrlib.branch.BranchHooks()
811
bzrlib.smart.server.SmartTCPServer.hooks = bzrlib.smart.server.SmartServerHooks()
813
def _silenceUI(self):
814
"""Turn off UI for duration of test"""
815
# by default the UI is off; tests can turn it on if they want it.
816
saved = ui.ui_factory
818
ui.ui_factory = saved
819
ui.ui_factory = ui.SilentUIFactory()
820
self.addCleanup(_restore)
822
def _ndiff_strings(self, a, b):
823
"""Return ndiff between two strings containing lines.
825
A trailing newline is added if missing to make the strings
827
if b and b[-1] != '\n':
829
if a and a[-1] != '\n':
831
difflines = difflib.ndiff(a.splitlines(True),
833
linejunk=lambda x: False,
834
charjunk=lambda x: False)
835
return ''.join(difflines)
837
def assertEqual(self, a, b, message=''):
841
except UnicodeError, e:
842
# If we can't compare without getting a UnicodeError, then
843
# obviously they are different
844
mutter('UnicodeError: %s', e)
847
raise AssertionError("%snot equal:\na = %s\nb = %s\n"
849
pformat(a), pformat(b)))
851
assertEquals = assertEqual
853
def assertEqualDiff(self, a, b, message=None):
854
"""Assert two texts are equal, if not raise an exception.
856
This is intended for use with multi-line strings where it can
857
be hard to find the differences by eye.
859
# TODO: perhaps override assertEquals to call this for strings?
863
message = "texts not equal:\n"
864
raise AssertionError(message +
865
self._ndiff_strings(a, b))
867
def assertEqualMode(self, mode, mode_test):
868
self.assertEqual(mode, mode_test,
869
'mode mismatch %o != %o' % (mode, mode_test))
871
def assertPositive(self, val):
872
"""Assert that val is greater than 0."""
873
self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
875
def assertNegative(self, val):
876
"""Assert that val is less than 0."""
877
self.assertTrue(val < 0, 'expected a negative value, but got %s' % val)
879
def assertStartsWith(self, s, prefix):
880
if not s.startswith(prefix):
881
raise AssertionError('string %r does not start with %r' % (s, prefix))
883
def assertEndsWith(self, s, suffix):
884
"""Asserts that s ends with suffix."""
885
if not s.endswith(suffix):
886
raise AssertionError('string %r does not end with %r' % (s, suffix))
888
def assertContainsRe(self, haystack, needle_re):
889
"""Assert that a contains something matching a regular expression."""
890
if not re.search(needle_re, haystack):
891
if '\n' in haystack or len(haystack) > 60:
892
# a long string, format it in a more readable way
893
raise AssertionError(
894
'pattern "%s" not found in\n"""\\\n%s"""\n'
895
% (needle_re, haystack))
897
raise AssertionError('pattern "%s" not found in "%s"'
898
% (needle_re, haystack))
900
def assertNotContainsRe(self, haystack, needle_re):
901
"""Assert that a does not match a regular expression"""
902
if re.search(needle_re, haystack):
903
raise AssertionError('pattern "%s" found in "%s"'
904
% (needle_re, haystack))
906
def assertSubset(self, sublist, superlist):
907
"""Assert that every entry in sublist is present in superlist."""
908
missing = set(sublist) - set(superlist)
910
raise AssertionError("value(s) %r not present in container %r" %
911
(missing, superlist))
913
def assertListRaises(self, excClass, func, *args, **kwargs):
914
"""Fail unless excClass is raised when the iterator from func is used.
916
Many functions can return generators this makes sure
917
to wrap them in a list() call to make sure the whole generator
918
is run, and that the proper exception is raised.
921
list(func(*args, **kwargs))
925
if getattr(excClass,'__name__', None) is not None:
926
excName = excClass.__name__
928
excName = str(excClass)
929
raise self.failureException, "%s not raised" % excName
931
def assertRaises(self, excClass, callableObj, *args, **kwargs):
932
"""Assert that a callable raises a particular exception.
934
:param excClass: As for the except statement, this may be either an
935
exception class, or a tuple of classes.
936
:param callableObj: A callable, will be passed ``*args`` and
939
Returns the exception so that you can examine it.
942
callableObj(*args, **kwargs)
946
if getattr(excClass,'__name__', None) is not None:
947
excName = excClass.__name__
950
excName = str(excClass)
951
raise self.failureException, "%s not raised" % excName
953
def assertIs(self, left, right, message=None):
954
if not (left is right):
955
if message is not None:
956
raise AssertionError(message)
958
raise AssertionError("%r is not %r." % (left, right))
960
def assertIsNot(self, left, right, message=None):
962
if message is not None:
963
raise AssertionError(message)
965
raise AssertionError("%r is %r." % (left, right))
967
def assertTransportMode(self, transport, path, mode):
968
"""Fail if a path does not have mode mode.
970
If modes are not supported on this transport, the assertion is ignored.
972
if not transport._can_roundtrip_unix_modebits():
974
path_stat = transport.stat(path)
975
actual_mode = stat.S_IMODE(path_stat.st_mode)
976
self.assertEqual(mode, actual_mode,
977
'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
979
def assertIsSameRealPath(self, path1, path2):
980
"""Fail if path1 and path2 points to different files"""
981
self.assertEqual(osutils.realpath(path1),
982
osutils.realpath(path2),
983
"apparent paths:\na = %s\nb = %s\n," % (path1, path2))
985
def assertIsInstance(self, obj, kls):
986
"""Fail if obj is not an instance of kls"""
987
if not isinstance(obj, kls):
988
self.fail("%r is an instance of %s rather than %s" % (
989
obj, obj.__class__, kls))
991
def expectFailure(self, reason, assertion, *args, **kwargs):
992
"""Invoke a test, expecting it to fail for the given reason.
994
This is for assertions that ought to succeed, but currently fail.
995
(The failure is *expected* but not *wanted*.) Please be very precise
996
about the failure you're expecting. If a new bug is introduced,
997
AssertionError should be raised, not KnownFailure.
999
Frequently, expectFailure should be followed by an opposite assertion.
1002
Intended to be used with a callable that raises AssertionError as the
1003
'assertion' parameter. args and kwargs are passed to the 'assertion'.
1005
Raises KnownFailure if the test fails. Raises AssertionError if the
1010
self.expectFailure('Math is broken', self.assertNotEqual, 54,
1012
self.assertEqual(42, dynamic_val)
1014
This means that a dynamic_val of 54 will cause the test to raise
1015
a KnownFailure. Once math is fixed and the expectFailure is removed,
1016
only a dynamic_val of 42 will allow the test to pass. Anything other
1017
than 54 or 42 will cause an AssertionError.
1020
assertion(*args, **kwargs)
1021
except AssertionError:
1022
raise KnownFailure(reason)
1024
self.fail('Unexpected success. Should have failed: %s' % reason)
1026
def assertFileEqual(self, content, path):
1027
"""Fail if path does not contain 'content'."""
1028
self.failUnlessExists(path)
1029
f = file(path, 'rb')
1034
self.assertEqualDiff(content, s)
1036
def failUnlessExists(self, path):
1037
"""Fail unless path or paths, which may be abs or relative, exist."""
1038
if not isinstance(path, basestring):
1040
self.failUnlessExists(p)
1042
self.failUnless(osutils.lexists(path),path+" does not exist")
1044
def failIfExists(self, path):
1045
"""Fail if path or paths, which may be abs or relative, exist."""
1046
if not isinstance(path, basestring):
1048
self.failIfExists(p)
1050
self.failIf(osutils.lexists(path),path+" exists")
1052
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1053
"""A helper for callDeprecated and applyDeprecated.
1055
:param a_callable: A callable to call.
1056
:param args: The positional arguments for the callable
1057
:param kwargs: The keyword arguments for the callable
1058
:return: A tuple (warnings, result). result is the result of calling
1059
a_callable(``*args``, ``**kwargs``).
1062
def capture_warnings(msg, cls=None, stacklevel=None):
1063
# we've hooked into a deprecation specific callpath,
1064
# only deprecations should getting sent via it.
1065
self.assertEqual(cls, DeprecationWarning)
1066
local_warnings.append(msg)
1067
original_warning_method = symbol_versioning.warn
1068
symbol_versioning.set_warning_method(capture_warnings)
1070
result = a_callable(*args, **kwargs)
1072
symbol_versioning.set_warning_method(original_warning_method)
1073
return (local_warnings, result)
1075
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
1076
"""Call a deprecated callable without warning the user.
1078
Note that this only captures warnings raised by symbol_versioning.warn,
1079
not other callers that go direct to the warning module.
1081
To test that a deprecated method raises an error, do something like
1084
self.assertRaises(errors.ReservedId,
1085
self.applyDeprecated,
1086
deprecated_in((1, 5, 0)),
1090
:param deprecation_format: The deprecation format that the callable
1091
should have been deprecated with. This is the same type as the
1092
parameter to deprecated_method/deprecated_function. If the
1093
callable is not deprecated with this format, an assertion error
1095
:param a_callable: A callable to call. This may be a bound method or
1096
a regular function. It will be called with ``*args`` and
1098
:param args: The positional arguments for the callable
1099
:param kwargs: The keyword arguments for the callable
1100
:return: The result of a_callable(``*args``, ``**kwargs``)
1102
call_warnings, result = self._capture_deprecation_warnings(a_callable,
1104
expected_first_warning = symbol_versioning.deprecation_string(
1105
a_callable, deprecation_format)
1106
if len(call_warnings) == 0:
1107
self.fail("No deprecation warning generated by call to %s" %
1109
self.assertEqual(expected_first_warning, call_warnings[0])
1112
def callCatchWarnings(self, fn, *args, **kw):
1113
"""Call a callable that raises python warnings.
1115
The caller's responsible for examining the returned warnings.
1117
If the callable raises an exception, the exception is not
1118
caught and propagates up to the caller. In that case, the list
1119
of warnings is not available.
1121
:returns: ([warning_object, ...], fn_result)
1123
# XXX: This is not perfect, because it completely overrides the
1124
# warnings filters, and some code may depend on suppressing particular
1125
# warnings. It's the easiest way to insulate ourselves from -Werror,
1126
# though. -- Andrew, 20071062
1128
def _catcher(message, category, filename, lineno, file=None):
1129
# despite the name, 'message' is normally(?) a Warning subclass
1131
wlist.append(message)
1132
saved_showwarning = warnings.showwarning
1133
saved_filters = warnings.filters
1135
warnings.showwarning = _catcher
1136
warnings.filters = []
1137
result = fn(*args, **kw)
1139
warnings.showwarning = saved_showwarning
1140
warnings.filters = saved_filters
1141
return wlist, result
1143
def callDeprecated(self, expected, callable, *args, **kwargs):
1144
"""Assert that a callable is deprecated in a particular way.
1146
This is a very precise test for unusual requirements. The
1147
applyDeprecated helper function is probably more suited for most tests
1148
as it allows you to simply specify the deprecation format being used
1149
and will ensure that that is issued for the function being called.
1151
Note that this only captures warnings raised by symbol_versioning.warn,
1152
not other callers that go direct to the warning module. To catch
1153
general warnings, use callCatchWarnings.
1155
:param expected: a list of the deprecation warnings expected, in order
1156
:param callable: The callable to call
1157
:param args: The positional arguments for the callable
1158
:param kwargs: The keyword arguments for the callable
1160
call_warnings, result = self._capture_deprecation_warnings(callable,
1162
self.assertEqual(expected, call_warnings)
1165
def _startLogFile(self):
1166
"""Send bzr and test log messages to a temporary file.
1168
The file is removed as the test is torn down.
1170
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
1171
self._log_file = os.fdopen(fileno, 'w+')
1172
self._log_memento = bzrlib.trace.push_log_file(self._log_file)
1173
self._log_file_name = name
1174
self.addCleanup(self._finishLogFile)
1176
def _finishLogFile(self):
1177
"""Finished with the log file.
1179
Close the file and delete it, unless setKeepLogfile was called.
1181
if self._log_file is None:
1183
bzrlib.trace.pop_log_file(self._log_memento)
1184
self._log_file.close()
1185
self._log_file = None
1186
if not self._keep_log_file:
1187
os.remove(self._log_file_name)
1188
self._log_file_name = None
1190
def setKeepLogfile(self):
1191
"""Make the logfile not be deleted when _finishLogFile is called."""
1192
self._keep_log_file = True
1194
def addCleanup(self, callable):
1195
"""Arrange to run a callable when this case is torn down.
1197
Callables are run in the reverse of the order they are registered,
1198
ie last-in first-out.
1200
if callable in self._cleanups:
1201
raise ValueError("cleanup function %r already registered on %s"
1203
self._cleanups.append(callable)
1205
def _cleanEnvironment(self):
1207
'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
1208
'HOME': os.getcwd(),
1209
'APPDATA': None, # bzr now use Win32 API and don't rely on APPDATA
1210
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
1212
'BZREMAIL': None, # may still be present in the environment
1214
'BZR_PROGRESS_BAR': None,
1217
'SSH_AUTH_SOCK': None,
1221
'https_proxy': None,
1222
'HTTPS_PROXY': None,
1227
# Nobody cares about these ones AFAIK. So far at
1228
# least. If you do (care), please update this comment
1232
'BZR_REMOTE_PATH': None,
1235
self.addCleanup(self._restoreEnvironment)
1236
for name, value in new_env.iteritems():
1237
self._captureVar(name, value)
1239
def _captureVar(self, name, newvalue):
1240
"""Set an environment variable, and reset it when finished."""
1241
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
1243
def _restore_debug_flags(self):
1244
debug.debug_flags.clear()
1245
debug.debug_flags.update(self._preserved_debug_flags)
1247
def _restoreEnvironment(self):
1248
for name, value in self.__old_env.iteritems():
1249
osutils.set_or_unset_env(name, value)
1251
def _restoreHooks(self):
1252
for klass, hooks in self._preserved_hooks.items():
1253
setattr(klass, 'hooks', hooks)
1255
def knownFailure(self, reason):
1256
"""This test has failed for some known reason."""
1257
raise KnownFailure(reason)
1259
def run(self, result=None):
1260
if result is None: result = self.defaultTestResult()
1261
for feature in getattr(self, '_test_needs_features', []):
1262
if not feature.available():
1263
result.startTest(self)
1264
if getattr(result, 'addNotSupported', None):
1265
result.addNotSupported(self, feature)
1267
result.addSuccess(self)
1268
result.stopTest(self)
1271
return unittest.TestCase.run(self, result)
1274
absent_attr = object()
1275
for attr_name in self.attrs_to_keep:
1276
attr = getattr(self, attr_name, absent_attr)
1277
if attr is not absent_attr:
1278
saved_attrs[attr_name] = attr
1279
self.__dict__ = saved_attrs
1283
unittest.TestCase.tearDown(self)
1285
def time(self, callable, *args, **kwargs):
1286
"""Run callable and accrue the time it takes to the benchmark time.
1288
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
1289
this will cause lsprofile statistics to be gathered and stored in
1292
if self._benchtime is None:
1296
if not self._gather_lsprof_in_benchmarks:
1297
return callable(*args, **kwargs)
1299
# record this benchmark
1300
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
1302
self._benchcalls.append(((callable, args, kwargs), stats))
1305
self._benchtime += time.time() - start
1307
def _runCleanups(self):
1308
"""Run registered cleanup functions.
1310
This should only be called from TestCase.tearDown.
1312
# TODO: Perhaps this should keep running cleanups even if
1313
# one of them fails?
1315
# Actually pop the cleanups from the list so tearDown running
1316
# twice is safe (this happens for skipped tests).
1317
while self._cleanups:
1318
self._cleanups.pop()()
1320
def log(self, *args):
1323
def _get_log(self, keep_log_file=False):
1324
"""Get the log from bzrlib.trace calls from this test.
1326
:param keep_log_file: When True, if the log is still a file on disk
1327
leave it as a file on disk. When False, if the log is still a file
1328
on disk, the log file is deleted and the log preserved as
1330
:return: A string containing the log.
1332
# flush the log file, to get all content
1334
bzrlib.trace._trace_file.flush()
1335
if self._log_contents:
1336
# XXX: this can hardly contain the content flushed above --vila
1338
return self._log_contents
1339
if self._log_file_name is not None:
1340
logfile = open(self._log_file_name)
1342
log_contents = logfile.read()
1345
if not keep_log_file:
1346
self._log_contents = log_contents
1348
os.remove(self._log_file_name)
1350
if sys.platform == 'win32' and e.errno == errno.EACCES:
1351
sys.stderr.write(('Unable to delete log file '
1352
' %r\n' % self._log_file_name))
1357
return "DELETED log file to reduce memory footprint"
1359
def requireFeature(self, feature):
1360
"""This test requires a specific feature is available.
1362
:raises UnavailableFeature: When feature is not available.
1364
if not feature.available():
1365
raise UnavailableFeature(feature)
1367
def _run_bzr_autosplit(self, args, retcode, encoding, stdin,
1369
"""Run bazaar command line, splitting up a string command line."""
1370
if isinstance(args, basestring):
1371
# shlex don't understand unicode strings,
1372
# so args should be plain string (bialix 20070906)
1373
args = list(shlex.split(str(args)))
1374
return self._run_bzr_core(args, retcode=retcode,
1375
encoding=encoding, stdin=stdin, working_dir=working_dir,
1378
def _run_bzr_core(self, args, retcode, encoding, stdin,
1380
if encoding is None:
1381
encoding = bzrlib.user_encoding
1382
stdout = StringIOWrapper()
1383
stderr = StringIOWrapper()
1384
stdout.encoding = encoding
1385
stderr.encoding = encoding
1387
self.log('run bzr: %r', args)
1388
# FIXME: don't call into logging here
1389
handler = logging.StreamHandler(stderr)
1390
handler.setLevel(logging.INFO)
1391
logger = logging.getLogger('')
1392
logger.addHandler(handler)
1393
old_ui_factory = ui.ui_factory
1394
ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)
1397
if working_dir is not None:
1398
cwd = osutils.getcwd()
1399
os.chdir(working_dir)
1402
result = self.apply_redirected(ui.ui_factory.stdin,
1404
bzrlib.commands.run_bzr_catch_user_errors,
1407
logger.removeHandler(handler)
1408
ui.ui_factory = old_ui_factory
1412
out = stdout.getvalue()
1413
err = stderr.getvalue()
1415
self.log('output:\n%r', out)
1417
self.log('errors:\n%r', err)
1418
if retcode is not None:
1419
self.assertEquals(retcode, result,
1420
message='Unexpected return code')
1423
def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
1424
working_dir=None, error_regexes=[], output_encoding=None):
1425
"""Invoke bzr, as if it were run from the command line.
1427
The argument list should not include the bzr program name - the
1428
first argument is normally the bzr command. Arguments may be
1429
passed in three ways:
1431
1- A list of strings, eg ["commit", "a"]. This is recommended
1432
when the command contains whitespace or metacharacters, or
1433
is built up at run time.
1435
2- A single string, eg "add a". This is the most convenient
1436
for hardcoded commands.
1438
This runs bzr through the interface that catches and reports
1439
errors, and with logging set to something approximating the
1440
default, so that error reporting can be checked.
1442
This should be the main method for tests that want to exercise the
1443
overall behavior of the bzr application (rather than a unit test
1444
or a functional test of the library.)
1446
This sends the stdout/stderr results into the test's log,
1447
where it may be useful for debugging. See also run_captured.
1449
:keyword stdin: A string to be used as stdin for the command.
1450
:keyword retcode: The status code the command should return;
1452
:keyword working_dir: The directory to run the command in
1453
:keyword error_regexes: A list of expected error messages. If
1454
specified they must be seen in the error output of the command.
1456
out, err = self._run_bzr_autosplit(
1461
working_dir=working_dir,
1463
for regex in error_regexes:
1464
self.assertContainsRe(err, regex)
1467
def run_bzr_error(self, error_regexes, *args, **kwargs):
1468
"""Run bzr, and check that stderr contains the supplied regexes
1470
:param error_regexes: Sequence of regular expressions which
1471
must each be found in the error output. The relative ordering
1473
:param args: command-line arguments for bzr
1474
:param kwargs: Keyword arguments which are interpreted by run_bzr
1475
This function changes the default value of retcode to be 3,
1476
since in most cases this is run when you expect bzr to fail.
1478
:return: (out, err) The actual output of running the command (in case
1479
you want to do more inspection)
1483
# Make sure that commit is failing because there is nothing to do
1484
self.run_bzr_error(['no changes to commit'],
1485
['commit', '-m', 'my commit comment'])
1486
# Make sure --strict is handling an unknown file, rather than
1487
# giving us the 'nothing to do' error
1488
self.build_tree(['unknown'])
1489
self.run_bzr_error(['Commit refused because there are unknown files'],
1490
['commit', --strict', '-m', 'my commit comment'])
1492
kwargs.setdefault('retcode', 3)
1493
kwargs['error_regexes'] = error_regexes
1494
out, err = self.run_bzr(*args, **kwargs)
1497
def run_bzr_subprocess(self, *args, **kwargs):
1498
"""Run bzr in a subprocess for testing.
1500
This starts a new Python interpreter and runs bzr in there.
1501
This should only be used for tests that have a justifiable need for
1502
this isolation: e.g. they are testing startup time, or signal
1503
handling, or early startup code, etc. Subprocess code can't be
1504
profiled or debugged so easily.
1506
:keyword retcode: The status code that is expected. Defaults to 0. If
1507
None is supplied, the status code is not checked.
1508
:keyword env_changes: A dictionary which lists changes to environment
1509
variables. A value of None will unset the env variable.
1510
The values must be strings. The change will only occur in the
1511
child, so you don't need to fix the environment after running.
1512
:keyword universal_newlines: Convert CRLF => LF
1513
:keyword allow_plugins: By default the subprocess is run with
1514
--no-plugins to ensure test reproducibility. Also, it is possible
1515
for system-wide plugins to create unexpected output on stderr,
1516
which can cause unnecessary test failures.
1518
env_changes = kwargs.get('env_changes', {})
1519
working_dir = kwargs.get('working_dir', None)
1520
allow_plugins = kwargs.get('allow_plugins', False)
1522
if isinstance(args[0], list):
1524
elif isinstance(args[0], basestring):
1525
args = list(shlex.split(args[0]))
1527
raise ValueError("passing varargs to run_bzr_subprocess")
1528
process = self.start_bzr_subprocess(args, env_changes=env_changes,
1529
working_dir=working_dir,
1530
allow_plugins=allow_plugins)
1531
# We distinguish between retcode=None and retcode not passed.
1532
supplied_retcode = kwargs.get('retcode', 0)
1533
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
1534
universal_newlines=kwargs.get('universal_newlines', False),
1537
def start_bzr_subprocess(self, process_args, env_changes=None,
1538
skip_if_plan_to_signal=False,
1540
allow_plugins=False):
1541
"""Start bzr in a subprocess for testing.
1543
This starts a new Python interpreter and runs bzr in there.
1544
This should only be used for tests that have a justifiable need for
1545
this isolation: e.g. they are testing startup time, or signal
1546
handling, or early startup code, etc. Subprocess code can't be
1547
profiled or debugged so easily.
1549
:param process_args: a list of arguments to pass to the bzr executable,
1550
for example ``['--version']``.
1551
:param env_changes: A dictionary which lists changes to environment
1552
variables. A value of None will unset the env variable.
1553
The values must be strings. The change will only occur in the
1554
child, so you don't need to fix the environment after running.
1555
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1557
:param allow_plugins: If False (default) pass --no-plugins to bzr.
1559
:returns: Popen object for the started process.
1561
if skip_if_plan_to_signal:
1562
if not getattr(os, 'kill', None):
1563
raise TestSkipped("os.kill not available.")
1565
if env_changes is None:
1569
def cleanup_environment():
1570
for env_var, value in env_changes.iteritems():
1571
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
1573
def restore_environment():
1574
for env_var, value in old_env.iteritems():
1575
osutils.set_or_unset_env(env_var, value)
1577
bzr_path = self.get_bzr_path()
1580
if working_dir is not None:
1581
cwd = osutils.getcwd()
1582
os.chdir(working_dir)
1585
# win32 subprocess doesn't support preexec_fn
1586
# so we will avoid using it on all platforms, just to
1587
# make sure the code path is used, and we don't break on win32
1588
cleanup_environment()
1589
command = [sys.executable, bzr_path]
1590
if not allow_plugins:
1591
command.append('--no-plugins')
1592
command.extend(process_args)
1593
process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1595
restore_environment()
1601
def _popen(self, *args, **kwargs):
1602
"""Place a call to Popen.
1604
Allows tests to override this method to intercept the calls made to
1605
Popen for introspection.
1607
return Popen(*args, **kwargs)
1609
def get_bzr_path(self):
1610
"""Return the path of the 'bzr' executable for this test suite."""
1611
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1612
if not os.path.isfile(bzr_path):
1613
# We are probably installed. Assume sys.argv is the right file
1614
bzr_path = sys.argv[0]
1617
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1618
universal_newlines=False, process_args=None):
1619
"""Finish the execution of process.
1621
:param process: the Popen object returned from start_bzr_subprocess.
1622
:param retcode: The status code that is expected. Defaults to 0. If
1623
None is supplied, the status code is not checked.
1624
:param send_signal: an optional signal to send to the process.
1625
:param universal_newlines: Convert CRLF => LF
1626
:returns: (stdout, stderr)
1628
if send_signal is not None:
1629
os.kill(process.pid, send_signal)
1630
out, err = process.communicate()
1632
if universal_newlines:
1633
out = out.replace('\r\n', '\n')
1634
err = err.replace('\r\n', '\n')
1636
if retcode is not None and retcode != process.returncode:
1637
if process_args is None:
1638
process_args = "(unknown args)"
1639
mutter('Output of bzr %s:\n%s', process_args, out)
1640
mutter('Error for bzr %s:\n%s', process_args, err)
1641
self.fail('Command bzr %s failed with retcode %s != %s'
1642
% (process_args, retcode, process.returncode))
1645
def check_inventory_shape(self, inv, shape):
1646
"""Compare an inventory to a list of expected names.
1648
Fail if they are not precisely equal.
1651
shape = list(shape) # copy
1652
for path, ie in inv.entries():
1653
name = path.replace('\\', '/')
1654
if ie.kind == 'directory':
1661
self.fail("expected paths not found in inventory: %r" % shape)
1663
self.fail("unexpected paths found in inventory: %r" % extras)
1665
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
1666
a_callable=None, *args, **kwargs):
1667
"""Call callable with redirected std io pipes.
1669
Returns the return code."""
1670
if not callable(a_callable):
1671
raise ValueError("a_callable must be callable.")
1673
stdin = StringIO("")
1675
if getattr(self, "_log_file", None) is not None:
1676
stdout = self._log_file
1680
if getattr(self, "_log_file", None is not None):
1681
stderr = self._log_file
1684
real_stdin = sys.stdin
1685
real_stdout = sys.stdout
1686
real_stderr = sys.stderr
1691
return a_callable(*args, **kwargs)
1693
sys.stdout = real_stdout
1694
sys.stderr = real_stderr
1695
sys.stdin = real_stdin
1697
def reduceLockdirTimeout(self):
1698
"""Reduce the default lock timeout for the duration of the test, so that
1699
if LockContention occurs during a test, it does so quickly.
1701
Tests that expect to provoke LockContention errors should call this.
1703
orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS
1705
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
1706
self.addCleanup(resetTimeout)
1707
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0
1709
def make_utf8_encoded_stringio(self, encoding_type=None):
1710
"""Return a StringIOWrapper instance, that will encode Unicode
1713
if encoding_type is None:
1714
encoding_type = 'strict'
1716
output_encoding = 'utf-8'
1717
sio = codecs.getwriter(output_encoding)(sio, errors=encoding_type)
1718
sio.encoding = output_encoding
1722
class TestCaseWithMemoryTransport(TestCase):
1723
"""Common test class for tests that do not need disk resources.
1725
Tests that need disk resources should derive from TestCaseWithTransport.
1727
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1729
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1730
a directory which does not exist. This serves to help ensure test isolation
1731
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1732
must exist. However, TestCaseWithMemoryTransport does not offer local
1733
file defaults for the transport in tests, nor does it obey the command line
1734
override, so tests that accidentally write to the common directory should
1737
:cvar TEST_ROOT: Directory containing all temporary directories, plus
1738
a .bzr directory that stops us ascending higher into the filesystem.
1744
def __init__(self, methodName='runTest'):
1745
# allow test parameterization after test construction and before test
1746
# execution. Variables that the parameterizer sets need to be
1747
# ones that are not set by setUp, or setUp will trash them.
1748
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1749
self.vfs_transport_factory = default_transport
1750
self.transport_server = None
1751
self.transport_readonly_server = None
1752
self.__vfs_server = None
1754
def get_transport(self, relpath=None):
1755
"""Return a writeable transport.
1757
This transport is for the test scratch space relative to
1760
:param relpath: a path relative to the base url.
1762
t = get_transport(self.get_url(relpath))
1763
self.assertFalse(t.is_readonly())
1766
def get_readonly_transport(self, relpath=None):
1767
"""Return a readonly transport for the test scratch space
1769
This can be used to test that operations which should only need
1770
readonly access in fact do not try to write.
1772
:param relpath: a path relative to the base url.
1774
t = get_transport(self.get_readonly_url(relpath))
1775
self.assertTrue(t.is_readonly())
1778
def create_transport_readonly_server(self):
1779
"""Create a transport server from class defined at init.
1781
This is mostly a hook for daughter classes.
1783
return self.transport_readonly_server()
1785
def get_readonly_server(self):
1786
"""Get the server instance for the readonly transport
1788
This is useful for some tests with specific servers to do diagnostics.
1790
if self.__readonly_server is None:
1791
if self.transport_readonly_server is None:
1792
# readonly decorator requested
1793
# bring up the server
1794
self.__readonly_server = ReadonlyServer()
1795
self.__readonly_server.setUp(self.get_vfs_only_server())
1797
self.__readonly_server = self.create_transport_readonly_server()
1798
self.__readonly_server.setUp(self.get_vfs_only_server())
1799
self.addCleanup(self.__readonly_server.tearDown)
1800
return self.__readonly_server
1802
def get_readonly_url(self, relpath=None):
1803
"""Get a URL for the readonly transport.
1805
This will either be backed by '.' or a decorator to the transport
1806
used by self.get_url()
1807
relpath provides for clients to get a path relative to the base url.
1808
These should only be downwards relative, not upwards.
1810
base = self.get_readonly_server().get_url()
1811
return self._adjust_url(base, relpath)
1813
def get_vfs_only_server(self):
1814
"""Get the vfs only read/write server instance.
1816
This is useful for some tests with specific servers that need
1819
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1820
is no means to override it.
1822
if self.__vfs_server is None:
1823
self.__vfs_server = MemoryServer()
1824
self.__vfs_server.setUp()
1825
self.addCleanup(self.__vfs_server.tearDown)
1826
return self.__vfs_server
1828
def get_server(self):
1829
"""Get the read/write server instance.
1831
This is useful for some tests with specific servers that need
1834
This is built from the self.transport_server factory. If that is None,
1835
then the self.get_vfs_server is returned.
1837
if self.__server is None:
1838
if self.transport_server is None or self.transport_server is self.vfs_transport_factory:
1839
return self.get_vfs_only_server()
1841
# bring up a decorated means of access to the vfs only server.
1842
self.__server = self.transport_server()
1844
self.__server.setUp(self.get_vfs_only_server())
1845
except TypeError, e:
1846
# This should never happen; the try:Except here is to assist
1847
# developers having to update code rather than seeing an
1848
# uninformative TypeError.
1849
raise Exception, "Old server API in use: %s, %s" % (self.__server, e)
1850
self.addCleanup(self.__server.tearDown)
1851
return self.__server
1853
def _adjust_url(self, base, relpath):
1854
"""Get a URL (or maybe a path) for the readwrite transport.
1856
This will either be backed by '.' or to an equivalent non-file based
1858
relpath provides for clients to get a path relative to the base url.
1859
These should only be downwards relative, not upwards.
1861
if relpath is not None and relpath != '.':
1862
if not base.endswith('/'):
1864
# XXX: Really base should be a url; we did after all call
1865
# get_url()! But sometimes it's just a path (from
1866
# LocalAbspathServer), and it'd be wrong to append urlescaped data
1867
# to a non-escaped local path.
1868
if base.startswith('./') or base.startswith('/'):
1871
base += urlutils.escape(relpath)
1874
def get_url(self, relpath=None):
1875
"""Get a URL (or maybe a path) for the readwrite transport.
1877
This will either be backed by '.' or to an equivalent non-file based
1879
relpath provides for clients to get a path relative to the base url.
1880
These should only be downwards relative, not upwards.
1882
base = self.get_server().get_url()
1883
return self._adjust_url(base, relpath)
1885
def get_vfs_only_url(self, relpath=None):
1886
"""Get a URL (or maybe a path for the plain old vfs transport.
1888
This will never be a smart protocol. It always has all the
1889
capabilities of the local filesystem, but it might actually be a
1890
MemoryTransport or some other similar virtual filesystem.
1892
This is the backing transport (if any) of the server returned by
1893
get_url and get_readonly_url.
1895
:param relpath: provides for clients to get a path relative to the base
1896
url. These should only be downwards relative, not upwards.
1899
base = self.get_vfs_only_server().get_url()
1900
return self._adjust_url(base, relpath)
1902
def _create_safety_net(self):
1903
"""Make a fake bzr directory.
1905
This prevents any tests propagating up onto the TEST_ROOT directory's
1908
root = TestCaseWithMemoryTransport.TEST_ROOT
1909
bzrdir.BzrDir.create_standalone_workingtree(root)
1911
def _check_safety_net(self):
1912
"""Check that the safety .bzr directory have not been touched.
1914
_make_test_root have created a .bzr directory to prevent tests from
1915
propagating. This method ensures than a test did not leaked.
1917
root = TestCaseWithMemoryTransport.TEST_ROOT
1918
wt = workingtree.WorkingTree.open(root)
1919
last_rev = wt.last_revision()
1920
if last_rev != 'null:':
1921
# The current test have modified the /bzr directory, we need to
1922
# recreate a new one or all the followng tests will fail.
1923
# If you need to inspect its content uncomment the following line
1924
# import pdb; pdb.set_trace()
1925
_rmtree_temp_dir(root + '/.bzr')
1926
self._create_safety_net()
1927
raise AssertionError('%s/.bzr should not be modified' % root)
1929
def _make_test_root(self):
1930
if TestCaseWithMemoryTransport.TEST_ROOT is None:
1931
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
1932
TestCaseWithMemoryTransport.TEST_ROOT = root
1934
self._create_safety_net()
1936
# The same directory is used by all tests, and we're not
1937
# specifically told when all tests are finished. This will do.
1938
atexit.register(_rmtree_temp_dir, root)
1940
self.addCleanup(self._check_safety_net)
1942
def makeAndChdirToTestDir(self):
1943
"""Create a temporary directories for this one test.
1945
This must set self.test_home_dir and self.test_dir and chdir to
1948
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1950
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1951
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1952
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1954
def make_branch(self, relpath, format=None):
1955
"""Create a branch on the transport at relpath."""
1956
repo = self.make_repository(relpath, format=format)
1957
return repo.bzrdir.create_branch()
1959
def make_bzrdir(self, relpath, format=None):
1961
# might be a relative or absolute path
1962
maybe_a_url = self.get_url(relpath)
1963
segments = maybe_a_url.rsplit('/', 1)
1964
t = get_transport(maybe_a_url)
1965
if len(segments) > 1 and segments[-1] not in ('', '.'):
1969
if isinstance(format, basestring):
1970
format = bzrdir.format_registry.make_bzrdir(format)
1971
return format.initialize_on_transport(t)
1972
except errors.UninitializableFormat:
1973
raise TestSkipped("Format %s is not initializable." % format)
1975
def make_repository(self, relpath, shared=False, format=None):
1976
"""Create a repository on our default transport at relpath.
1978
Note that relpath must be a relative path, not a full url.
1980
# FIXME: If you create a remoterepository this returns the underlying
1981
# real format, which is incorrect. Actually we should make sure that
1982
# RemoteBzrDir returns a RemoteRepository.
1983
# maybe mbp 20070410
1984
made_control = self.make_bzrdir(relpath, format=format)
1985
return made_control.create_repository(shared=shared)
1987
def make_branch_and_memory_tree(self, relpath, format=None):
1988
"""Create a branch on the default transport and a MemoryTree for it."""
1989
b = self.make_branch(relpath, format=format)
1990
return memorytree.MemoryTree.create_on_branch(b)
1992
def overrideEnvironmentForTesting(self):
1993
os.environ['HOME'] = self.test_home_dir
1994
os.environ['BZR_HOME'] = self.test_home_dir
1997
super(TestCaseWithMemoryTransport, self).setUp()
1998
self._make_test_root()
1999
_currentdir = os.getcwdu()
2000
def _leaveDirectory():
2001
os.chdir(_currentdir)
2002
self.addCleanup(_leaveDirectory)
2003
self.makeAndChdirToTestDir()
2004
self.overrideEnvironmentForTesting()
2005
self.__readonly_server = None
2006
self.__server = None
2007
self.reduceLockdirTimeout()
2010
class TestCaseInTempDir(TestCaseWithMemoryTransport):
2011
"""Derived class that runs a test within a temporary directory.
2013
This is useful for tests that need to create a branch, etc.
2015
The directory is created in a slightly complex way: for each
2016
Python invocation, a new temporary top-level directory is created.
2017
All test cases create their own directory within that. If the
2018
tests complete successfully, the directory is removed.
2020
:ivar test_base_dir: The path of the top-level directory for this
2021
test, which contains a home directory and a work directory.
2023
:ivar test_home_dir: An initially empty directory under test_base_dir
2024
which is used as $HOME for this test.
2026
:ivar test_dir: A directory under test_base_dir used as the current
2027
directory when the test proper is run.
2030
OVERRIDE_PYTHON = 'python'
2032
def check_file_contents(self, filename, expect):
2033
self.log("check contents of file %s" % filename)
2034
contents = file(filename, 'r').read()
2035
if contents != expect:
2036
self.log("expected: %r" % expect)
2037
self.log("actually: %r" % contents)
2038
self.fail("contents of %s not as expected" % filename)
2040
def makeAndChdirToTestDir(self):
2041
"""See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
2043
For TestCaseInTempDir we create a temporary directory based on the test
2044
name and then create two subdirs - test and home under it.
2046
# create a directory within the top level test directory
2047
candidate_dir = osutils.mkdtemp(dir=self.TEST_ROOT)
2048
# now create test and home directories within this dir
2049
self.test_base_dir = candidate_dir
2050
self.test_home_dir = self.test_base_dir + '/home'
2051
os.mkdir(self.test_home_dir)
2052
self.test_dir = self.test_base_dir + '/work'
2053
os.mkdir(self.test_dir)
2054
os.chdir(self.test_dir)
2055
# put name of test inside
2056
f = file(self.test_base_dir + '/name', 'w')
2061
self.addCleanup(self.deleteTestDir)
2063
def deleteTestDir(self):
2064
os.chdir(self.TEST_ROOT)
2065
_rmtree_temp_dir(self.test_base_dir)
2067
def build_tree(self, shape, line_endings='binary', transport=None):
2068
"""Build a test tree according to a pattern.
2070
shape is a sequence of file specifications. If the final
2071
character is '/', a directory is created.
2073
This assumes that all the elements in the tree being built are new.
2075
This doesn't add anything to a branch.
2077
:type shape: list or tuple.
2078
:param line_endings: Either 'binary' or 'native'
2079
in binary mode, exact contents are written in native mode, the
2080
line endings match the default platform endings.
2081
:param transport: A transport to write to, for building trees on VFS's.
2082
If the transport is readonly or None, "." is opened automatically.
2085
if type(shape) not in (list, tuple):
2086
raise AssertionError("Parameter 'shape' should be "
2087
"a list or a tuple. Got %r instead" % (shape,))
2088
# It's OK to just create them using forward slashes on windows.
2089
if transport is None or transport.is_readonly():
2090
transport = get_transport(".")
2092
self.assert_(isinstance(name, basestring))
2094
transport.mkdir(urlutils.escape(name[:-1]))
2096
if line_endings == 'binary':
2098
elif line_endings == 'native':
2101
raise errors.BzrError(
2102
'Invalid line ending request %r' % line_endings)
2103
content = "contents of %s%s" % (name.encode('utf-8'), end)
2104
transport.put_bytes_non_atomic(urlutils.escape(name), content)
2106
def build_tree_contents(self, shape):
2107
build_tree_contents(shape)
2109
def assertInWorkingTree(self, path, root_path='.', tree=None):
2110
"""Assert whether path or paths are in the WorkingTree"""
2112
tree = workingtree.WorkingTree.open(root_path)
2113
if not isinstance(path, basestring):
2115
self.assertInWorkingTree(p,tree=tree)
2117
self.assertIsNot(tree.path2id(path), None,
2118
path+' not in working tree.')
2120
def assertNotInWorkingTree(self, path, root_path='.', tree=None):
2121
"""Assert whether path or paths are not in the WorkingTree"""
2123
tree = workingtree.WorkingTree.open(root_path)
2124
if not isinstance(path, basestring):
2126
self.assertNotInWorkingTree(p,tree=tree)
2128
self.assertIs(tree.path2id(path), None, path+' in working tree.')
2131
class TestCaseWithTransport(TestCaseInTempDir):
2132
"""A test case that provides get_url and get_readonly_url facilities.
2134
These back onto two transport servers, one for readonly access and one for
2137
If no explicit class is provided for readonly access, a
2138
ReadonlyTransportDecorator is used instead which allows the use of non disk
2139
based read write transports.
2141
If an explicit class is provided for readonly access, that server and the
2142
readwrite one must both define get_url() as resolving to os.getcwd().
2145
def get_vfs_only_server(self):
2146
"""See TestCaseWithMemoryTransport.
2148
This is useful for some tests with specific servers that need
2151
if self.__vfs_server is None:
2152
self.__vfs_server = self.vfs_transport_factory()
2153
self.__vfs_server.setUp()
2154
self.addCleanup(self.__vfs_server.tearDown)
2155
return self.__vfs_server
2157
def make_branch_and_tree(self, relpath, format=None):
2158
"""Create a branch on the transport and a tree locally.
2160
If the transport is not a LocalTransport, the Tree can't be created on
2161
the transport. In that case if the vfs_transport_factory is
2162
LocalURLServer the working tree is created in the local
2163
directory backing the transport, and the returned tree's branch and
2164
repository will also be accessed locally. Otherwise a lightweight
2165
checkout is created and returned.
2167
:param format: The BzrDirFormat.
2168
:returns: the WorkingTree.
2170
# TODO: always use the local disk path for the working tree,
2171
# this obviously requires a format that supports branch references
2172
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
2174
b = self.make_branch(relpath, format=format)
2176
return b.bzrdir.create_workingtree()
2177
except errors.NotLocalUrl:
2178
# We can only make working trees locally at the moment. If the
2179
# transport can't support them, then we keep the non-disk-backed
2180
# branch and create a local checkout.
2181
if self.vfs_transport_factory is LocalURLServer:
2182
# the branch is colocated on disk, we cannot create a checkout.
2183
# hopefully callers will expect this.
2184
local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
2185
return local_controldir.create_workingtree()
2187
return b.create_checkout(relpath, lightweight=True)
2189
def assertIsDirectory(self, relpath, transport):
2190
"""Assert that relpath within transport is a directory.
2192
This may not be possible on all transports; in that case it propagates
2193
a TransportNotPossible.
2196
mode = transport.stat(relpath).st_mode
2197
except errors.NoSuchFile:
2198
self.fail("path %s is not a directory; no such file"
2200
if not stat.S_ISDIR(mode):
2201
self.fail("path %s is not a directory; has mode %#o"
2204
def assertTreesEqual(self, left, right):
2205
"""Check that left and right have the same content and properties."""
2206
# we use a tree delta to check for equality of the content, and we
2207
# manually check for equality of other things such as the parents list.
2208
self.assertEqual(left.get_parent_ids(), right.get_parent_ids())
2209
differences = left.changes_from(right)
2210
self.assertFalse(differences.has_changed(),
2211
"Trees %r and %r are different: %r" % (left, right, differences))
2214
super(TestCaseWithTransport, self).setUp()
2215
self.__vfs_server = None
2218
class ChrootedTestCase(TestCaseWithTransport):
2219
"""A support class that provides readonly urls outside the local namespace.
2221
This is done by checking if self.transport_server is a MemoryServer. if it
2222
is then we are chrooted already, if it is not then an HttpServer is used
2225
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
2226
be used without needed to redo it when a different
2227
subclass is in use ?
2231
super(ChrootedTestCase, self).setUp()
2232
if not self.vfs_transport_factory == MemoryServer:
2233
self.transport_readonly_server = HttpServer
2236
def condition_id_re(pattern):
2237
"""Create a condition filter which performs a re check on a test's id.
2239
:param pattern: A regular expression string.
2240
:return: A callable that returns True if the re matches.
2242
filter_re = re.compile(pattern)
2243
def condition(test):
2245
return filter_re.search(test_id)
2249
def condition_isinstance(klass_or_klass_list):
2250
"""Create a condition filter which returns isinstance(param, klass).
2252
:return: A callable which when called with one parameter obj return the
2253
result of isinstance(obj, klass_or_klass_list).
2256
return isinstance(obj, klass_or_klass_list)
2260
def condition_id_in_list(id_list):
2261
"""Create a condition filter which verify that test's id in a list.
2263
:param id_list: A TestIdList object.
2264
:return: A callable that returns True if the test's id appears in the list.
2266
def condition(test):
2267
return id_list.includes(test.id())
2271
def exclude_tests_by_condition(suite, condition):
2272
"""Create a test suite which excludes some tests from suite.
2274
:param suite: The suite to get tests from.
2275
:param condition: A callable whose result evaluates True when called with a
2276
test case which should be excluded from the result.
2277
:return: A suite which contains the tests found in suite that fail
2281
for test in iter_suite_tests(suite):
2282
if not condition(test):
2284
return TestUtil.TestSuite(result)
2287
def filter_suite_by_condition(suite, condition):
2288
"""Create a test suite by filtering another one.
2290
:param suite: The source suite.
2291
:param condition: A callable whose result evaluates True when called with a
2292
test case which should be included in the result.
2293
:return: A suite which contains the tests found in suite that pass
2297
for test in iter_suite_tests(suite):
2300
return TestUtil.TestSuite(result)
2303
def filter_suite_by_re(suite, pattern):
2304
"""Create a test suite by filtering another one.
2306
:param suite: the source suite
2307
:param pattern: pattern that names must match
2308
:returns: the newly created suite
2310
condition = condition_id_re(pattern)
2311
result_suite = filter_suite_by_condition(suite, condition)
2315
def filter_suite_by_id_list(suite, test_id_list):
2316
"""Create a test suite by filtering another one.
2318
:param suite: The source suite.
2319
:param test_id_list: A list of the test ids to keep as strings.
2320
:returns: the newly created suite
2322
condition = condition_id_in_list(test_id_list)
2323
result_suite = filter_suite_by_condition(suite, condition)
2327
def exclude_tests_by_re(suite, pattern):
2328
"""Create a test suite which excludes some tests from suite.
2330
:param suite: The suite to get tests from.
2331
:param pattern: A regular expression string. Test ids that match this
2332
pattern will be excluded from the result.
2333
:return: A TestSuite that contains all the tests from suite without the
2334
tests that matched pattern. The order of tests is the same as it was in
2337
return exclude_tests_by_condition(suite, condition_id_re(pattern))
2340
def preserve_input(something):
2341
"""A helper for performing test suite transformation chains.
2343
:param something: Anything you want to preserve.
2349
def randomize_suite(suite):
2350
"""Return a new TestSuite with suite's tests in random order.
2352
The tests in the input suite are flattened into a single suite in order to
2353
accomplish this. Any nested TestSuites are removed to provide global
2356
tests = list(iter_suite_tests(suite))
2357
random.shuffle(tests)
2358
return TestUtil.TestSuite(tests)
2361
def split_suite_by_condition(suite, condition):
2362
"""Split a test suite into two by a condition.
2364
:param suite: The suite to split.
2365
:param condition: The condition to match on. Tests that match this
2366
condition are returned in the first test suite, ones that do not match
2367
are in the second suite.
2368
:return: A tuple of two test suites, where the first contains tests from
2369
suite matching the condition, and the second contains the remainder
2370
from suite. The order within each output suite is the same as it was in
2375
for test in iter_suite_tests(suite):
2377
matched.append(test)
2379
did_not_match.append(test)
2380
return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
2383
def split_suite_by_re(suite, pattern):
2384
"""Split a test suite into two by a regular expression.
2386
:param suite: The suite to split.
2387
:param pattern: A regular expression string. Test ids that match this
2388
pattern will be in the first test suite returned, and the others in the
2389
second test suite returned.
2390
:return: A tuple of two test suites, where the first contains tests from
2391
suite matching pattern, and the second contains the remainder from
2392
suite. The order within each output suite is the same as it was in
2395
return split_suite_by_condition(suite, condition_id_re(pattern))
2398
def run_suite(suite, name='test', verbose=False, pattern=".*",
2399
stop_on_failure=False,
2400
transport=None, lsprof_timed=None, bench_history=None,
2401
matching_tests_first=None,
2404
exclude_pattern=None,
2406
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
2411
runner = TextTestRunner(stream=sys.stdout,
2413
verbosity=verbosity,
2414
bench_history=bench_history,
2415
list_only=list_only,
2417
runner.stop_on_failure=stop_on_failure
2418
# Initialise the random number generator and display the seed used.
2419
# We convert the seed to a long to make it reuseable across invocations.
2420
random_order = False
2421
if random_seed is not None:
2423
if random_seed == "now":
2424
random_seed = long(time.time())
2426
# Convert the seed to a long if we can
2428
random_seed = long(random_seed)
2431
runner.stream.writeln("Randomizing test order using seed %s\n" %
2433
random.seed(random_seed)
2434
# Customise the list of tests if requested
2435
if exclude_pattern is not None:
2436
suite = exclude_tests_by_re(suite, exclude_pattern)
2438
order_changer = randomize_suite
2440
order_changer = preserve_input
2441
if pattern != '.*' or random_order:
2442
if matching_tests_first:
2443
suites = map(order_changer, split_suite_by_re(suite, pattern))
2444
suite = TestUtil.TestSuite(suites)
2446
suite = order_changer(filter_suite_by_re(suite, pattern))
2448
result = runner.run(suite)
2451
return result.wasStrictlySuccessful()
2453
return result.wasSuccessful()
2456
# Controlled by "bzr selftest -E=..." option
2457
selftest_debug_flags = set()
2460
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
2462
test_suite_factory=None,
2465
matching_tests_first=None,
2468
exclude_pattern=None,
2473
"""Run the whole test suite under the enhanced runner"""
2474
# XXX: Very ugly way to do this...
2475
# Disable warning about old formats because we don't want it to disturb
2476
# any blackbox tests.
2477
from bzrlib import repository
2478
repository._deprecation_warning_done = True
2480
global default_transport
2481
if transport is None:
2482
transport = default_transport
2483
old_transport = default_transport
2484
default_transport = transport
2485
global selftest_debug_flags
2486
old_debug_flags = selftest_debug_flags
2487
if debug_flags is not None:
2488
selftest_debug_flags = set(debug_flags)
2490
if load_list is None:
2493
keep_only = load_test_id_list(load_list)
2494
if test_suite_factory is None:
2495
suite = test_suite(keep_only)
2497
suite = test_suite_factory()
2498
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
2499
stop_on_failure=stop_on_failure,
2500
transport=transport,
2501
lsprof_timed=lsprof_timed,
2502
bench_history=bench_history,
2503
matching_tests_first=matching_tests_first,
2504
list_only=list_only,
2505
random_seed=random_seed,
2506
exclude_pattern=exclude_pattern,
2509
default_transport = old_transport
2510
selftest_debug_flags = old_debug_flags
2513
def load_test_id_list(file_name):
2514
"""Load a test id list from a text file.
2516
The format is one test id by line. No special care is taken to impose
2517
strict rules, these test ids are used to filter the test suite so a test id
2518
that do not match an existing test will do no harm. This allows user to add
2519
comments, leave blank lines, etc.
2523
ftest = open(file_name, 'rt')
2525
if e.errno != errno.ENOENT:
2528
raise errors.NoSuchFile(file_name)
2530
for test_name in ftest.readlines():
2531
test_list.append(test_name.strip())
2536
def suite_matches_id_list(test_suite, id_list):
2537
"""Warns about tests not appearing or appearing more than once.
2539
:param test_suite: A TestSuite object.
2540
:param test_id_list: The list of test ids that should be found in
2543
:return: (absents, duplicates) absents is a list containing the test found
2544
in id_list but not in test_suite, duplicates is a list containing the
2545
test found multiple times in test_suite.
2547
When using a prefined test id list, it may occurs that some tests do not
2548
exist anymore or that some tests use the same id. This function warns the
2549
tester about potential problems in his workflow (test lists are volatile)
2550
or in the test suite itself (using the same id for several tests does not
2551
help to localize defects).
2553
# Build a dict counting id occurrences
2555
for test in iter_suite_tests(test_suite):
2557
tests[id] = tests.get(id, 0) + 1
2562
occurs = tests.get(id, 0)
2564
not_found.append(id)
2566
duplicates.append(id)
2568
return not_found, duplicates
2571
class TestIdList(object):
2572
"""Test id list to filter a test suite.
2574
Relying on the assumption that test ids are built as:
2575
<module>[.<class>.<method>][(<param>+)], <module> being in python dotted
2576
notation, this class offers methods to :
2577
- avoid building a test suite for modules not refered to in the test list,
2578
- keep only the tests listed from the module test suite.
2581
def __init__(self, test_id_list):
2582
# When a test suite needs to be filtered against us we compare test ids
2583
# for equality, so a simple dict offers a quick and simple solution.
2584
self.tests = dict().fromkeys(test_id_list, True)
2586
# While unittest.TestCase have ids like:
2587
# <module>.<class>.<method>[(<param+)],
2588
# doctest.DocTestCase can have ids like:
2591
# <module>.<function>
2592
# <module>.<class>.<method>
2594
# Since we can't predict a test class from its name only, we settle on
2595
# a simple constraint: a test id always begins with its module name.
2598
for test_id in test_id_list:
2599
parts = test_id.split('.')
2600
mod_name = parts.pop(0)
2601
modules[mod_name] = True
2603
mod_name += '.' + part
2604
modules[mod_name] = True
2605
self.modules = modules
2607
def refers_to(self, module_name):
2608
"""Is there tests for the module or one of its sub modules."""
2609
return self.modules.has_key(module_name)
2611
def includes(self, test_id):
2612
return self.tests.has_key(test_id)
2615
def test_suite(keep_only=None):
2616
"""Build and return TestSuite for the whole of bzrlib.
2618
:param keep_only: A list of test ids limiting the suite returned.
2620
This function can be replaced if you need to change the default test
2621
suite on a global basis, but it is not encouraged.
2625
'bzrlib.util.tests.test_bencode',
2626
'bzrlib.tests.blackbox',
2627
'bzrlib.tests.branch_implementations',
2628
'bzrlib.tests.bzrdir_implementations',
2629
'bzrlib.tests.commands',
2630
'bzrlib.tests.inventory_implementations',
2631
'bzrlib.tests.interrepository_implementations',
2632
'bzrlib.tests.intertree_implementations',
2633
'bzrlib.tests.interversionedfile_implementations',
2634
'bzrlib.tests.per_lock',
2635
'bzrlib.tests.repository_implementations',
2636
'bzrlib.tests.revisionstore_implementations',
2637
'bzrlib.tests.test__dirstate_helpers',
2638
'bzrlib.tests.test_ancestry',
2639
'bzrlib.tests.test_annotate',
2640
'bzrlib.tests.test_api',
2641
'bzrlib.tests.test_atomicfile',
2642
'bzrlib.tests.test_bad_files',
2643
'bzrlib.tests.test_bisect_multi',
2644
'bzrlib.tests.test_branch',
2645
'bzrlib.tests.test_branchbuilder',
2646
'bzrlib.tests.test_bugtracker',
2647
'bzrlib.tests.test_bundle',
2648
'bzrlib.tests.test_bzrdir',
2649
'bzrlib.tests.test_cache_utf8',
2650
'bzrlib.tests.test_commands',
2651
'bzrlib.tests.test_commit',
2652
'bzrlib.tests.test_commit_merge',
2653
'bzrlib.tests.test_config',
2654
'bzrlib.tests.test_conflicts',
2655
'bzrlib.tests.test_counted_lock',
2656
'bzrlib.tests.test_decorators',
2657
'bzrlib.tests.test_delta',
2658
'bzrlib.tests.test_deprecated_graph',
2659
'bzrlib.tests.test_diff',
2660
'bzrlib.tests.test_dirstate',
2661
'bzrlib.tests.test_directory_service',
2662
'bzrlib.tests.test_email_message',
2663
'bzrlib.tests.test_errors',
2664
'bzrlib.tests.test_escaped_store',
2665
'bzrlib.tests.test_extract',
2666
'bzrlib.tests.test_fetch',
2667
'bzrlib.tests.test_ftp_transport',
2668
'bzrlib.tests.test_generate_docs',
2669
'bzrlib.tests.test_generate_ids',
2670
'bzrlib.tests.test_globbing',
2671
'bzrlib.tests.test_gpg',
2672
'bzrlib.tests.test_graph',
2673
'bzrlib.tests.test_hashcache',
2674
'bzrlib.tests.test_help',
2675
'bzrlib.tests.test_hooks',
2676
'bzrlib.tests.test_http',
2677
'bzrlib.tests.test_http_implementations',
2678
'bzrlib.tests.test_http_response',
2679
'bzrlib.tests.test_https_ca_bundle',
2680
'bzrlib.tests.test_identitymap',
2681
'bzrlib.tests.test_ignores',
2682
'bzrlib.tests.test_index',
2683
'bzrlib.tests.test_info',
2684
'bzrlib.tests.test_inv',
2685
'bzrlib.tests.test_knit',
2686
'bzrlib.tests.test_lazy_import',
2687
'bzrlib.tests.test_lazy_regex',
2688
'bzrlib.tests.test_lockdir',
2689
'bzrlib.tests.test_lockable_files',
2690
'bzrlib.tests.test_log',
2691
'bzrlib.tests.test_lsprof',
2692
'bzrlib.tests.test_lru_cache',
2693
'bzrlib.tests.test_mail_client',
2694
'bzrlib.tests.test_memorytree',
2695
'bzrlib.tests.test_merge',
2696
'bzrlib.tests.test_merge3',
2697
'bzrlib.tests.test_merge_core',
2698
'bzrlib.tests.test_merge_directive',
2699
'bzrlib.tests.test_missing',
2700
'bzrlib.tests.test_msgeditor',
2701
'bzrlib.tests.test_multiparent',
2702
'bzrlib.tests.test_mutabletree',
2703
'bzrlib.tests.test_nonascii',
2704
'bzrlib.tests.test_options',
2705
'bzrlib.tests.test_osutils',
2706
'bzrlib.tests.test_osutils_encodings',
2707
'bzrlib.tests.test_pack',
2708
'bzrlib.tests.test_patch',
2709
'bzrlib.tests.test_patches',
2710
'bzrlib.tests.test_permissions',
2711
'bzrlib.tests.test_plugins',
2712
'bzrlib.tests.test_progress',
2713
'bzrlib.tests.test_read_bundle',
2714
'bzrlib.tests.test_reconfigure',
2715
'bzrlib.tests.test_reconcile',
2716
'bzrlib.tests.test_registry',
2717
'bzrlib.tests.test_remote',
2718
'bzrlib.tests.test_repository',
2719
'bzrlib.tests.test_revert',
2720
'bzrlib.tests.test_revision',
2721
'bzrlib.tests.test_revisionspec',
2722
'bzrlib.tests.test_revisiontree',
2723
'bzrlib.tests.test_rio',
2724
'bzrlib.tests.test_sampler',
2725
'bzrlib.tests.test_selftest',
2726
'bzrlib.tests.test_setup',
2727
'bzrlib.tests.test_sftp_transport',
2728
'bzrlib.tests.test_smart',
2729
'bzrlib.tests.test_smart_add',
2730
'bzrlib.tests.test_smart_transport',
2731
'bzrlib.tests.test_smtp_connection',
2732
'bzrlib.tests.test_source',
2733
'bzrlib.tests.test_ssh_transport',
2734
'bzrlib.tests.test_status',
2735
'bzrlib.tests.test_store',
2736
'bzrlib.tests.test_strace',
2737
'bzrlib.tests.test_subsume',
2738
'bzrlib.tests.test_switch',
2739
'bzrlib.tests.test_symbol_versioning',
2740
'bzrlib.tests.test_tag',
2741
'bzrlib.tests.test_testament',
2742
'bzrlib.tests.test_textfile',
2743
'bzrlib.tests.test_textmerge',
2744
'bzrlib.tests.test_timestamp',
2745
'bzrlib.tests.test_trace',
2746
'bzrlib.tests.test_transactions',
2747
'bzrlib.tests.test_transform',
2748
'bzrlib.tests.test_transport',
2749
'bzrlib.tests.test_transport_implementations',
2750
'bzrlib.tests.test_tree',
2751
'bzrlib.tests.test_treebuilder',
2752
'bzrlib.tests.test_tsort',
2753
'bzrlib.tests.test_tuned_gzip',
2754
'bzrlib.tests.test_ui',
2755
'bzrlib.tests.test_uncommit',
2756
'bzrlib.tests.test_upgrade',
2757
'bzrlib.tests.test_urlutils',
2758
'bzrlib.tests.test_versionedfile',
2759
'bzrlib.tests.test_version',
2760
'bzrlib.tests.test_version_info',
2761
'bzrlib.tests.test_weave',
2762
'bzrlib.tests.test_whitebox',
2763
'bzrlib.tests.test_win32utils',
2764
'bzrlib.tests.test_workingtree',
2765
'bzrlib.tests.test_workingtree_4',
2766
'bzrlib.tests.test_wsgi',
2767
'bzrlib.tests.test_xml',
2768
'bzrlib.tests.tree_implementations',
2769
'bzrlib.tests.workingtree_implementations',
2772
loader = TestUtil.TestLoader()
2774
if keep_only is None:
2775
loader = TestUtil.TestLoader()
2777
id_filter = TestIdList(keep_only)
2778
loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
2779
suite = loader.suiteClass()
2781
# modules building their suite with loadTestsFromModuleNames
2782
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
2784
modules_to_doctest = [
2789
'bzrlib.iterablefile',
2794
'bzrlib.symbol_versioning',
2797
'bzrlib.version_info_formats.format_custom',
2800
for mod in modules_to_doctest:
2801
if not (keep_only is None or id_filter.refers_to(mod)):
2802
# No tests to keep here, move along
2805
doc_suite = doctest.DocTestSuite(mod)
2806
except ValueError, e:
2807
print '**failed to get doctest for: %s\n%s' % (mod, e)
2809
suite.addTest(doc_suite)
2811
default_encoding = sys.getdefaultencoding()
2812
for name, plugin in bzrlib.plugin.plugins().items():
2813
if keep_only is not None:
2814
if not id_filter.refers_to(plugin.module.__name__):
2816
plugin_suite = plugin.test_suite()
2817
# We used to catch ImportError here and turn it into just a warning,
2818
# but really if you don't have --no-plugins this should be a failure.
2819
# mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
2820
if plugin_suite is None:
2821
plugin_suite = plugin.load_plugin_tests(loader)
2822
if plugin_suite is not None:
2823
suite.addTest(plugin_suite)
2824
if default_encoding != sys.getdefaultencoding():
2825
bzrlib.trace.warning(
2826
'Plugin "%s" tried to reset default encoding to: %s', name,
2827
sys.getdefaultencoding())
2829
sys.setdefaultencoding(default_encoding)
2831
if keep_only is not None:
2832
# Now that the referred modules have loaded their tests, keep only the
2834
suite = filter_suite_by_id_list(suite, id_filter)
2835
# Do some sanity checks on the id_list filtering
2836
not_found, duplicates = suite_matches_id_list(suite, keep_only)
2837
for id in not_found:
2838
bzrlib.trace.warning('"%s" not found in the test suite', id)
2839
for id in duplicates:
2840
bzrlib.trace.warning('"%s" is used as an id by several tests', id)
2845
def multiply_tests_from_modules(module_name_list, scenario_iter, loader=None):
2846
"""Adapt all tests in some given modules to given scenarios.
2848
This is the recommended public interface for test parameterization.
2849
Typically the test_suite() method for a per-implementation test
2850
suite will call multiply_tests_from_modules and return the
2853
:param module_name_list: List of fully-qualified names of test
2855
:param scenario_iter: Iterable of pairs of (scenario_name,
2856
scenario_param_dict).
2857
:param loader: If provided, will be used instead of a new
2858
bzrlib.tests.TestLoader() instance.
2860
This returns a new TestSuite containing the cross product of
2861
all the tests in all the modules, each repeated for each scenario.
2862
Each test is adapted by adding the scenario name at the end
2863
of its name, and updating the test object's __dict__ with the
2864
scenario_param_dict.
2866
>>> r = multiply_tests_from_modules(
2867
... ['bzrlib.tests.test_sampler'],
2868
... [('one', dict(param=1)),
2869
... ('two', dict(param=2))])
2870
>>> tests = list(iter_suite_tests(r))
2874
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
2880
# XXX: Isn't load_tests() a better way to provide the same functionality
2881
# without forcing a predefined TestScenarioApplier ? --vila 080215
2883
loader = TestUtil.TestLoader()
2885
suite = loader.suiteClass()
2887
adapter = TestScenarioApplier()
2888
adapter.scenarios = list(scenario_iter)
2889
adapt_modules(module_name_list, adapter, loader, suite)
2893
def multiply_scenarios(scenarios_left, scenarios_right):
2894
"""Multiply two sets of scenarios.
2896
:returns: the cartesian product of the two sets of scenarios, that is
2897
a scenario for every possible combination of a left scenario and a
2901
('%s,%s' % (left_name, right_name),
2902
dict(left_dict.items() + right_dict.items()))
2903
for left_name, left_dict in scenarios_left
2904
for right_name, right_dict in scenarios_right]
2908
def adapt_modules(mods_list, adapter, loader, suite):
2909
"""Adapt the modules in mods_list using adapter and add to suite."""
2910
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
2911
suite.addTests(adapter.adapt(test))
2914
def adapt_tests(tests_list, adapter, loader, suite):
2915
"""Adapt the tests in tests_list using adapter and add to suite."""
2916
for test in tests_list:
2917
suite.addTests(adapter.adapt(loader.loadTestsFromName(test)))
2920
def _rmtree_temp_dir(dirname):
2921
# If LANG=C we probably have created some bogus paths
2922
# which rmtree(unicode) will fail to delete
2923
# so make sure we are using rmtree(str) to delete everything
2924
# except on win32, where rmtree(str) will fail
2925
# since it doesn't have the property of byte-stream paths
2926
# (they are either ascii or mbcs)
2927
if sys.platform == 'win32':
2928
# make sure we are using the unicode win32 api
2929
dirname = unicode(dirname)
2931
dirname = dirname.encode(sys.getfilesystemencoding())
2933
osutils.rmtree(dirname)
2935
if sys.platform == 'win32' and e.errno == errno.EACCES:
2936
sys.stderr.write(('Permission denied: '
2937
'unable to remove testing dir '
2938
'%s\n' % os.path.basename(dirname)))
2943
class Feature(object):
2944
"""An operating system Feature."""
2947
self._available = None
2949
def available(self):
2950
"""Is the feature available?
2952
:return: True if the feature is available.
2954
if self._available is None:
2955
self._available = self._probe()
2956
return self._available
2959
"""Implement this method in concrete features.
2961
:return: True if the feature is available.
2963
raise NotImplementedError
2966
if getattr(self, 'feature_name', None):
2967
return self.feature_name()
2968
return self.__class__.__name__
2971
class _SymlinkFeature(Feature):
2974
return osutils.has_symlinks()
2976
def feature_name(self):
2979
SymlinkFeature = _SymlinkFeature()
2982
class _HardlinkFeature(Feature):
2985
return osutils.has_hardlinks()
2987
def feature_name(self):
2990
HardlinkFeature = _HardlinkFeature()
2993
class _OsFifoFeature(Feature):
2996
return getattr(os, 'mkfifo', None)
2998
def feature_name(self):
2999
return 'filesystem fifos'
3001
OsFifoFeature = _OsFifoFeature()
3004
class TestScenarioApplier(object):
3005
"""A tool to apply scenarios to tests."""
3007
def adapt(self, test):
3008
"""Return a TestSuite containing a copy of test for each scenario."""
3009
result = unittest.TestSuite()
3010
for scenario in self.scenarios:
3011
result.addTest(self.adapt_test_to_scenario(test, scenario))
3014
def adapt_test_to_scenario(self, test, scenario):
3015
"""Copy test and apply scenario to it.
3017
:param test: A test to adapt.
3018
:param scenario: A tuple describing the scenarion.
3019
The first element of the tuple is the new test id.
3020
The second element is a dict containing attributes to set on the
3022
:return: The adapted test.
3024
from copy import deepcopy
3025
new_test = deepcopy(test)
3026
for name, value in scenario[1].items():
3027
setattr(new_test, name, value)
3028
new_id = "%s(%s)" % (new_test.id(), scenario[0])
3029
new_test.id = lambda: new_id
3033
def probe_unicode_in_user_encoding():
3034
"""Try to encode several unicode strings to use in unicode-aware tests.
3035
Return first successfull match.
3037
:return: (unicode value, encoded plain string value) or (None, None)
3039
possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
3040
for uni_val in possible_vals:
3042
str_val = uni_val.encode(bzrlib.user_encoding)
3043
except UnicodeEncodeError:
3044
# Try a different character
3047
return uni_val, str_val
3051
def probe_bad_non_ascii(encoding):
3052
"""Try to find [bad] character with code [128..255]
3053
that cannot be decoded to unicode in some encoding.
3054
Return None if all non-ascii characters is valid
3057
for i in xrange(128, 256):
3060
char.decode(encoding)
3061
except UnicodeDecodeError:
3066
class _FTPServerFeature(Feature):
3067
"""Some tests want an FTP Server, check if one is available.
3069
Right now, the only way this is available is if 'medusa' is installed.
3070
http://www.amk.ca/python/code/medusa.html
3075
import bzrlib.tests.ftp_server
3080
def feature_name(self):
3083
FTPServerFeature = _FTPServerFeature()
3086
class _CaseInsensitiveFilesystemFeature(Feature):
3087
"""Check if underlined filesystem is case-insensitive
3088
(e.g. on Windows, Cygwin, MacOS)
3092
if TestCaseWithMemoryTransport.TEST_ROOT is None:
3093
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
3094
TestCaseWithMemoryTransport.TEST_ROOT = root
3096
root = TestCaseWithMemoryTransport.TEST_ROOT
3097
tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
3099
name_a = osutils.pathjoin(tdir, 'a')
3100
name_A = osutils.pathjoin(tdir, 'A')
3102
result = osutils.isdir(name_A)
3103
_rmtree_temp_dir(tdir)
3106
def feature_name(self):
3107
return 'case-insensitive filesystem'
3109
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
added
removed
bzrlib/tests/__init__.py
