1
# Copyright (C) 2005, 2006 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.
30
from cStringIO import StringIO
39
from subprocess import Popen, PIPE
56
import bzrlib.commands
57
import bzrlib.bundle.serializer
59
import bzrlib.inventory
60
import bzrlib.iterablefile
65
# lsprof not available
67
from bzrlib.merge import merge_inner
71
from bzrlib.revision import common_ancestor
73
from bzrlib import symbol_versioning
75
from bzrlib.transport import get_transport
76
import bzrlib.transport
77
from bzrlib.transport.local import LocalURLServer
78
from bzrlib.transport.memory import MemoryServer
79
from bzrlib.transport.readonly import ReadonlyServer
80
from bzrlib.trace import mutter, note
81
from bzrlib.tests import TestUtil
82
from bzrlib.tests.HttpServer import HttpServer
83
from bzrlib.tests.TestUtil import (
87
from bzrlib.tests.treeshape import build_tree_contents
88
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
90
default_transport = LocalURLServer
93
MODULES_TO_DOCTEST = [
94
bzrlib.bundle.serializer,
106
def packages_to_test():
107
"""Return a list of packages to test.
109
The packages are not globally imported so that import failures are
110
triggered when running selftest, not when importing the command.
113
import bzrlib.tests.blackbox
114
import bzrlib.tests.branch_implementations
115
import bzrlib.tests.bzrdir_implementations
116
import bzrlib.tests.interrepository_implementations
117
import bzrlib.tests.interversionedfile_implementations
118
import bzrlib.tests.intertree_implementations
119
import bzrlib.tests.repository_implementations
120
import bzrlib.tests.revisionstore_implementations
121
import bzrlib.tests.tree_implementations
122
import bzrlib.tests.workingtree_implementations
125
bzrlib.tests.blackbox,
126
bzrlib.tests.branch_implementations,
127
bzrlib.tests.bzrdir_implementations,
128
bzrlib.tests.interrepository_implementations,
129
bzrlib.tests.interversionedfile_implementations,
130
bzrlib.tests.intertree_implementations,
131
bzrlib.tests.repository_implementations,
132
bzrlib.tests.revisionstore_implementations,
133
bzrlib.tests.tree_implementations,
134
bzrlib.tests.workingtree_implementations,
138
class ExtendedTestResult(unittest._TextTestResult):
139
"""Accepts, reports and accumulates the results of running tests.
141
Compared to this unittest version this class adds support for profiling,
142
benchmarking, stopping as soon as a test fails, and skipping tests.
143
There are further-specialized subclasses for different types of display.
148
def __init__(self, stream, descriptions, verbosity,
152
"""Construct new TestResult.
154
:param bench_history: Optionally, a writable file object to accumulate
157
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
158
if bench_history is not None:
159
from bzrlib.version import _get_bzr_source_tree
160
src_tree = _get_bzr_source_tree()
163
revision_id = src_tree.get_parent_ids()[0]
165
# XXX: if this is a brand new tree, do the same as if there
169
# XXX: If there's no branch, what should we do?
171
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
172
self._bench_history = bench_history
173
self.ui = bzrlib.ui.ui_factory
174
self.num_tests = num_tests
176
self.failure_count = 0
179
self._overall_start_time = time.time()
181
def extractBenchmarkTime(self, testCase):
182
"""Add a benchmark time for the current test case."""
183
self._benchmarkTime = getattr(testCase, "_benchtime", None)
185
def _elapsedTestTimeString(self):
186
"""Return a time string for the overall time the current test has taken."""
187
return self._formatTime(time.time() - self._start_time)
189
def _testTimeString(self):
190
if self._benchmarkTime is not None:
192
self._formatTime(self._benchmarkTime),
193
self._elapsedTestTimeString())
195
return " %s" % self._elapsedTestTimeString()
197
def _formatTime(self, seconds):
198
"""Format seconds as milliseconds with leading spaces."""
199
return "%5dms" % (1000 * seconds)
201
def _shortened_test_description(self, test):
203
what = re.sub(r'^bzrlib\.(tests|benchmark)\.', '', what)
206
def startTest(self, test):
207
unittest.TestResult.startTest(self, test)
208
self.report_test_start(test)
209
self._recordTestStartTime()
211
def _recordTestStartTime(self):
212
"""Record that a test has started."""
213
self._start_time = time.time()
215
def addError(self, test, err):
216
if isinstance(err[1], TestSkipped):
217
return self.addSkipped(test, err)
218
unittest.TestResult.addError(self, test, err)
219
# We can only do this if we have one of our TestCases, not if
221
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
222
if setKeepLogfile is not None:
224
self.extractBenchmarkTime(test)
225
self.report_error(test, err)
229
def addFailure(self, test, err):
230
unittest.TestResult.addFailure(self, test, err)
231
# We can only do this if we have one of our TestCases, not if
233
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
234
if setKeepLogfile is not None:
236
self.extractBenchmarkTime(test)
237
self.report_failure(test, err)
241
def addSuccess(self, test):
242
self.extractBenchmarkTime(test)
243
if self._bench_history is not None:
244
if self._benchmarkTime is not None:
245
self._bench_history.write("%s %s\n" % (
246
self._formatTime(self._benchmarkTime),
248
self.report_success(test)
249
unittest.TestResult.addSuccess(self, test)
251
def addSkipped(self, test, skip_excinfo):
252
self.extractBenchmarkTime(test)
253
self.report_skip(test, skip_excinfo)
254
# seems best to treat this as success from point-of-view of unittest
255
# -- it actually does nothing so it barely matters :)
258
except KeyboardInterrupt:
261
self.addError(test, test.__exc_info())
263
unittest.TestResult.addSuccess(self, test)
265
def printErrorList(self, flavour, errors):
266
for test, err in errors:
267
self.stream.writeln(self.separator1)
268
self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
269
if getattr(test, '_get_log', None) is not None:
271
print >>self.stream, \
272
('vvvv[log from %s]' % test.id()).ljust(78,'-')
273
print >>self.stream, test._get_log()
274
print >>self.stream, \
275
('^^^^[log from %s]' % test.id()).ljust(78,'-')
276
self.stream.writeln(self.separator2)
277
self.stream.writeln("%s" % err)
282
def report_cleaning_up(self):
285
def report_success(self, test):
289
class TextTestResult(ExtendedTestResult):
290
"""Displays progress and results of tests in text form"""
292
def __init__(self, *args, **kw):
293
ExtendedTestResult.__init__(self, *args, **kw)
294
self.pb = self.ui.nested_progress_bar()
295
self.pb.show_pct = False
296
self.pb.show_spinner = False
297
self.pb.show_eta = False,
298
self.pb.show_count = False
299
self.pb.show_bar = False
301
def report_starting(self):
302
self.pb.update('[test 0/%d] starting...' % (self.num_tests))
304
def _progress_prefix_text(self):
305
a = '[%d' % self.count
306
if self.num_tests is not None:
307
a +='/%d' % self.num_tests
308
a += ' in %ds' % (time.time() - self._overall_start_time)
310
a += ', %d errors' % self.error_count
311
if self.failure_count:
312
a += ', %d failed' % self.failure_count
314
a += ', %d skipped' % self.skip_count
318
def report_test_start(self, test):
321
self._progress_prefix_text()
323
+ self._shortened_test_description(test))
325
def report_error(self, test, err):
326
self.error_count += 1
327
self.pb.note('ERROR: %s\n %s\n',
328
self._shortened_test_description(test),
332
def report_failure(self, test, err):
333
self.failure_count += 1
334
self.pb.note('FAIL: %s\n %s\n',
335
self._shortened_test_description(test),
339
def report_skip(self, test, skip_excinfo):
342
# at the moment these are mostly not things we can fix
343
# and so they just produce stipple; use the verbose reporter
346
# show test and reason for skip
347
self.pb.note('SKIP: %s\n %s\n',
348
self._shortened_test_description(test),
351
# since the class name was left behind in the still-visible
353
self.pb.note('SKIP: %s', skip_excinfo[1])
355
def report_cleaning_up(self):
356
self.pb.update('cleaning up...')
362
class VerboseTestResult(ExtendedTestResult):
363
"""Produce long output, with one line per test run plus times"""
365
def _ellipsize_to_right(self, a_string, final_width):
366
"""Truncate and pad a string, keeping the right hand side"""
367
if len(a_string) > final_width:
368
result = '...' + a_string[3-final_width:]
371
return result.ljust(final_width)
373
def report_starting(self):
374
self.stream.write('running %d tests...\n' % self.num_tests)
376
def report_test_start(self, test):
378
name = self._shortened_test_description(test)
379
self.stream.write(self._ellipsize_to_right(name,
380
osutils.terminal_width()-20))
383
def report_error(self, test, err):
384
self.error_count += 1
385
self.stream.writeln('ERROR %s\n %s'
386
% (self._testTimeString(), err[1]))
388
def report_failure(self, test, err):
389
self.failure_count += 1
390
self.stream.writeln(' FAIL %s\n %s'
391
% (self._testTimeString(), err[1]))
393
def report_success(self, test):
394
self.stream.writeln(' OK %s' % self._testTimeString())
395
for bench_called, stats in getattr(test, '_benchcalls', []):
396
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
397
stats.pprint(file=self.stream)
400
def report_skip(self, test, skip_excinfo):
401
print >>self.stream, ' SKIP %s' % self._testTimeString()
402
print >>self.stream, ' %s' % skip_excinfo[1]
405
class TextTestRunner(object):
406
stop_on_failure = False
414
self.stream = unittest._WritelnDecorator(stream)
415
self.descriptions = descriptions
416
self.verbosity = verbosity
417
self.keep_output = keep_output
418
self._bench_history = bench_history
421
"Run the given test case or test suite."
422
startTime = time.time()
423
if self.verbosity == 1:
424
result_class = TextTestResult
425
elif self.verbosity >= 2:
426
result_class = VerboseTestResult
427
result = result_class(self.stream,
430
bench_history=self._bench_history,
431
num_tests=test.countTestCases(),
433
result.stop_early = self.stop_on_failure
434
result.report_starting()
436
stopTime = time.time()
437
timeTaken = stopTime - startTime
439
self.stream.writeln(result.separator2)
440
run = result.testsRun
441
self.stream.writeln("Ran %d test%s in %.3fs" %
442
(run, run != 1 and "s" or "", timeTaken))
443
self.stream.writeln()
444
if not result.wasSuccessful():
445
self.stream.write("FAILED (")
446
failed, errored = map(len, (result.failures, result.errors))
448
self.stream.write("failures=%d" % failed)
450
if failed: self.stream.write(", ")
451
self.stream.write("errors=%d" % errored)
452
self.stream.writeln(")")
454
self.stream.writeln("OK")
455
result.report_cleaning_up()
456
# This is still a little bogus,
457
# but only a little. Folk not using our testrunner will
458
# have to delete their temp directories themselves.
459
test_root = TestCaseWithMemoryTransport.TEST_ROOT
460
if result.wasSuccessful() or not self.keep_output:
461
if test_root is not None:
462
# If LANG=C we probably have created some bogus paths
463
# which rmtree(unicode) will fail to delete
464
# so make sure we are using rmtree(str) to delete everything
465
# except on win32, where rmtree(str) will fail
466
# since it doesn't have the property of byte-stream paths
467
# (they are either ascii or mbcs)
468
if sys.platform == 'win32':
469
# make sure we are using the unicode win32 api
470
test_root = unicode(test_root)
472
test_root = test_root.encode(
473
sys.getfilesystemencoding())
474
osutils.rmtree(test_root)
476
note("Failed tests working directories are in '%s'\n", test_root)
477
TestCaseWithMemoryTransport.TEST_ROOT = None
482
def iter_suite_tests(suite):
483
"""Return all tests in a suite, recursing through nested suites"""
484
for item in suite._tests:
485
if isinstance(item, unittest.TestCase):
487
elif isinstance(item, unittest.TestSuite):
488
for r in iter_suite_tests(item):
491
raise Exception('unknown object %r inside test suite %r'
495
class TestSkipped(Exception):
496
"""Indicates that a test was intentionally skipped, rather than failing."""
499
class CommandFailed(Exception):
503
class StringIOWrapper(object):
504
"""A wrapper around cStringIO which just adds an encoding attribute.
506
Internally we can check sys.stdout to see what the output encoding
507
should be. However, cStringIO has no encoding attribute that we can
508
set. So we wrap it instead.
513
def __init__(self, s=None):
515
self.__dict__['_cstring'] = StringIO(s)
517
self.__dict__['_cstring'] = StringIO()
519
def __getattr__(self, name, getattr=getattr):
520
return getattr(self.__dict__['_cstring'], name)
522
def __setattr__(self, name, val):
523
if name == 'encoding':
524
self.__dict__['encoding'] = val
526
return setattr(self._cstring, name, val)
529
class TestCase(unittest.TestCase):
530
"""Base class for bzr unit tests.
532
Tests that need access to disk resources should subclass
533
TestCaseInTempDir not TestCase.
535
Error and debug log messages are redirected from their usual
536
location into a temporary file, the contents of which can be
537
retrieved by _get_log(). We use a real OS file, not an in-memory object,
538
so that it can also capture file IO. When the test completes this file
539
is read into memory and removed from disk.
541
There are also convenience functions to invoke bzr's command-line
542
routine, and to build and check bzr trees.
544
In addition to the usual method of overriding tearDown(), this class also
545
allows subclasses to register functions into the _cleanups list, which is
546
run in order as the object is torn down. It's less likely this will be
547
accidentally overlooked.
550
_log_file_name = None
552
_keep_log_file = False
553
# record lsprof data when performing benchmark calls.
554
_gather_lsprof_in_benchmarks = False
556
def __init__(self, methodName='testMethod'):
557
super(TestCase, self).__init__(methodName)
561
unittest.TestCase.setUp(self)
562
self._cleanEnvironment()
563
bzrlib.trace.disable_default_logging()
566
self._benchcalls = []
567
self._benchtime = None
569
def _silenceUI(self):
570
"""Turn off UI for duration of test"""
571
# by default the UI is off; tests can turn it on if they want it.
572
saved = bzrlib.ui.ui_factory
574
bzrlib.ui.ui_factory = saved
575
bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
576
self.addCleanup(_restore)
578
def _ndiff_strings(self, a, b):
579
"""Return ndiff between two strings containing lines.
581
A trailing newline is added if missing to make the strings
583
if b and b[-1] != '\n':
585
if a and a[-1] != '\n':
587
difflines = difflib.ndiff(a.splitlines(True),
589
linejunk=lambda x: False,
590
charjunk=lambda x: False)
591
return ''.join(difflines)
593
def assertEqualDiff(self, a, b, message=None):
594
"""Assert two texts are equal, if not raise an exception.
596
This is intended for use with multi-line strings where it can
597
be hard to find the differences by eye.
599
# TODO: perhaps override assertEquals to call this for strings?
603
message = "texts not equal:\n"
604
raise AssertionError(message +
605
self._ndiff_strings(a, b))
607
def assertEqualMode(self, mode, mode_test):
608
self.assertEqual(mode, mode_test,
609
'mode mismatch %o != %o' % (mode, mode_test))
611
def assertStartsWith(self, s, prefix):
612
if not s.startswith(prefix):
613
raise AssertionError('string %r does not start with %r' % (s, prefix))
615
def assertEndsWith(self, s, suffix):
616
"""Asserts that s ends with suffix."""
617
if not s.endswith(suffix):
618
raise AssertionError('string %r does not end with %r' % (s, suffix))
620
def assertContainsRe(self, haystack, needle_re):
621
"""Assert that a contains something matching a regular expression."""
622
if not re.search(needle_re, haystack):
623
raise AssertionError('pattern "%s" not found in "%s"'
624
% (needle_re, haystack))
626
def assertNotContainsRe(self, haystack, needle_re):
627
"""Assert that a does not match a regular expression"""
628
if re.search(needle_re, haystack):
629
raise AssertionError('pattern "%s" found in "%s"'
630
% (needle_re, haystack))
632
def assertSubset(self, sublist, superlist):
633
"""Assert that every entry in sublist is present in superlist."""
635
for entry in sublist:
636
if entry not in superlist:
637
missing.append(entry)
639
raise AssertionError("value(s) %r not present in container %r" %
640
(missing, superlist))
642
def assertListRaises(self, excClass, func, *args, **kwargs):
643
"""Fail unless excClass is raised when the iterator from func is used.
645
Many functions can return generators this makes sure
646
to wrap them in a list() call to make sure the whole generator
647
is run, and that the proper exception is raised.
650
list(func(*args, **kwargs))
654
if getattr(excClass,'__name__', None) is not None:
655
excName = excClass.__name__
657
excName = str(excClass)
658
raise self.failureException, "%s not raised" % excName
660
def assertIs(self, left, right):
661
if not (left is right):
662
raise AssertionError("%r is not %r." % (left, right))
664
def assertTransportMode(self, transport, path, mode):
665
"""Fail if a path does not have mode mode.
667
If modes are not supported on this transport, the assertion is ignored.
669
if not transport._can_roundtrip_unix_modebits():
671
path_stat = transport.stat(path)
672
actual_mode = stat.S_IMODE(path_stat.st_mode)
673
self.assertEqual(mode, actual_mode,
674
'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
676
def assertIsInstance(self, obj, kls):
677
"""Fail if obj is not an instance of kls"""
678
if not isinstance(obj, kls):
679
self.fail("%r is an instance of %s rather than %s" % (
680
obj, obj.__class__, kls))
682
def _capture_warnings(self, a_callable, *args, **kwargs):
683
"""A helper for callDeprecated and applyDeprecated.
685
:param a_callable: A callable to call.
686
:param args: The positional arguments for the callable
687
:param kwargs: The keyword arguments for the callable
688
:return: A tuple (warnings, result). result is the result of calling
689
a_callable(*args, **kwargs).
692
def capture_warnings(msg, cls=None, stacklevel=None):
693
# we've hooked into a deprecation specific callpath,
694
# only deprecations should getting sent via it.
695
self.assertEqual(cls, DeprecationWarning)
696
local_warnings.append(msg)
697
original_warning_method = symbol_versioning.warn
698
symbol_versioning.set_warning_method(capture_warnings)
700
result = a_callable(*args, **kwargs)
702
symbol_versioning.set_warning_method(original_warning_method)
703
return (local_warnings, result)
705
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
706
"""Call a deprecated callable without warning the user.
708
:param deprecation_format: The deprecation format that the callable
709
should have been deprecated with. This is the same type as the
710
parameter to deprecated_method/deprecated_function. If the
711
callable is not deprecated with this format, an assertion error
713
:param a_callable: A callable to call. This may be a bound method or
714
a regular function. It will be called with *args and **kwargs.
715
:param args: The positional arguments for the callable
716
:param kwargs: The keyword arguments for the callable
717
:return: The result of a_callable(*args, **kwargs)
719
call_warnings, result = self._capture_warnings(a_callable,
721
expected_first_warning = symbol_versioning.deprecation_string(
722
a_callable, deprecation_format)
723
if len(call_warnings) == 0:
724
self.fail("No assertion generated by call to %s" %
726
self.assertEqual(expected_first_warning, call_warnings[0])
729
def callDeprecated(self, expected, callable, *args, **kwargs):
730
"""Assert that a callable is deprecated in a particular way.
732
This is a very precise test for unusual requirements. The
733
applyDeprecated helper function is probably more suited for most tests
734
as it allows you to simply specify the deprecation format being used
735
and will ensure that that is issued for the function being called.
737
:param expected: a list of the deprecation warnings expected, in order
738
:param callable: The callable to call
739
:param args: The positional arguments for the callable
740
:param kwargs: The keyword arguments for the callable
742
call_warnings, result = self._capture_warnings(callable,
744
self.assertEqual(expected, call_warnings)
747
def _startLogFile(self):
748
"""Send bzr and test log messages to a temporary file.
750
The file is removed as the test is torn down.
752
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
753
self._log_file = os.fdopen(fileno, 'w+')
754
self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
755
self._log_file_name = name
756
self.addCleanup(self._finishLogFile)
758
def _finishLogFile(self):
759
"""Finished with the log file.
761
Close the file and delete it, unless setKeepLogfile was called.
763
if self._log_file is None:
765
bzrlib.trace.disable_test_log(self._log_nonce)
766
self._log_file.close()
767
self._log_file = None
768
if not self._keep_log_file:
769
os.remove(self._log_file_name)
770
self._log_file_name = None
772
def setKeepLogfile(self):
773
"""Make the logfile not be deleted when _finishLogFile is called."""
774
self._keep_log_file = True
776
def addCleanup(self, callable):
777
"""Arrange to run a callable when this case is torn down.
779
Callables are run in the reverse of the order they are registered,
780
ie last-in first-out.
782
if callable in self._cleanups:
783
raise ValueError("cleanup function %r already registered on %s"
785
self._cleanups.append(callable)
787
def _cleanEnvironment(self):
789
'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
791
'APPDATA': os.getcwd(),
793
'BZREMAIL': None, # may still be present in the environment
795
'BZR_PROGRESS_BAR': None,
805
# Nobody cares about these ones AFAIK. So far at
806
# least. If you do (care), please update this comment
812
self.addCleanup(self._restoreEnvironment)
813
for name, value in new_env.iteritems():
814
self._captureVar(name, value)
816
def _captureVar(self, name, newvalue):
817
"""Set an environment variable, and reset it when finished."""
818
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
820
def _restoreEnvironment(self):
821
for name, value in self.__old_env.iteritems():
822
osutils.set_or_unset_env(name, value)
826
unittest.TestCase.tearDown(self)
828
def time(self, callable, *args, **kwargs):
829
"""Run callable and accrue the time it takes to the benchmark time.
831
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
832
this will cause lsprofile statistics to be gathered and stored in
835
if self._benchtime is None:
839
if not self._gather_lsprof_in_benchmarks:
840
return callable(*args, **kwargs)
842
# record this benchmark
843
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
845
self._benchcalls.append(((callable, args, kwargs), stats))
848
self._benchtime += time.time() - start
850
def _runCleanups(self):
851
"""Run registered cleanup functions.
853
This should only be called from TestCase.tearDown.
855
# TODO: Perhaps this should keep running cleanups even if
857
for cleanup_fn in reversed(self._cleanups):
860
def log(self, *args):
863
def _get_log(self, keep_log_file=False):
864
"""Return as a string the log for this test. If the file is still
865
on disk and keep_log_file=False, delete the log file and store the
866
content in self._log_contents."""
867
# flush the log file, to get all content
869
bzrlib.trace._trace_file.flush()
870
if self._log_contents:
871
return self._log_contents
872
if self._log_file_name is not None:
873
logfile = open(self._log_file_name)
875
log_contents = logfile.read()
878
if not keep_log_file:
879
self._log_contents = log_contents
880
os.remove(self._log_file_name)
883
return "DELETED log file to reduce memory footprint"
885
def capture(self, cmd, retcode=0):
886
"""Shortcut that splits cmd into words, runs, and returns stdout"""
887
return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
889
def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None,
891
"""Invoke bzr and return (stdout, stderr).
893
Useful for code that wants to check the contents of the
894
output, the way error messages are presented, etc.
896
This should be the main method for tests that want to exercise the
897
overall behavior of the bzr application (rather than a unit test
898
or a functional test of the library.)
900
Much of the old code runs bzr by forking a new copy of Python, but
901
that is slower, harder to debug, and generally not necessary.
903
This runs bzr through the interface that catches and reports
904
errors, and with logging set to something approximating the
905
default, so that error reporting can be checked.
907
:param argv: arguments to invoke bzr
908
:param retcode: expected return code, or None for don't-care.
909
:param encoding: encoding for sys.stdout and sys.stderr
910
:param stdin: A string to be used as stdin for the command.
911
:param working_dir: Change to this directory before running
914
encoding = bzrlib.user_encoding
915
if stdin is not None:
916
stdin = StringIO(stdin)
917
stdout = StringIOWrapper()
918
stderr = StringIOWrapper()
919
stdout.encoding = encoding
920
stderr.encoding = encoding
922
self.log('run bzr: %r', argv)
923
# FIXME: don't call into logging here
924
handler = logging.StreamHandler(stderr)
925
handler.setLevel(logging.INFO)
926
logger = logging.getLogger('')
927
logger.addHandler(handler)
928
old_ui_factory = bzrlib.ui.ui_factory
929
bzrlib.ui.ui_factory = bzrlib.tests.blackbox.TestUIFactory(
932
bzrlib.ui.ui_factory.stdin = stdin
935
if working_dir is not None:
936
cwd = osutils.getcwd()
937
os.chdir(working_dir)
940
saved_debug_flags = frozenset(debug.debug_flags)
941
debug.debug_flags.clear()
943
result = self.apply_redirected(stdin, stdout, stderr,
944
bzrlib.commands.run_bzr_catch_errors,
947
debug.debug_flags.update(saved_debug_flags)
949
logger.removeHandler(handler)
950
bzrlib.ui.ui_factory = old_ui_factory
954
out = stdout.getvalue()
955
err = stderr.getvalue()
957
self.log('output:\n%r', out)
959
self.log('errors:\n%r', err)
960
if retcode is not None:
961
self.assertEquals(retcode, result)
964
def run_bzr(self, *args, **kwargs):
965
"""Invoke bzr, as if it were run from the command line.
967
This should be the main method for tests that want to exercise the
968
overall behavior of the bzr application (rather than a unit test
969
or a functional test of the library.)
971
This sends the stdout/stderr results into the test's log,
972
where it may be useful for debugging. See also run_captured.
974
:param stdin: A string to be used as stdin for the command.
976
retcode = kwargs.pop('retcode', 0)
977
encoding = kwargs.pop('encoding', None)
978
stdin = kwargs.pop('stdin', None)
979
working_dir = kwargs.pop('working_dir', None)
980
return self.run_bzr_captured(args, retcode=retcode, encoding=encoding,
981
stdin=stdin, working_dir=working_dir)
983
def run_bzr_decode(self, *args, **kwargs):
984
if 'encoding' in kwargs:
985
encoding = kwargs['encoding']
987
encoding = bzrlib.user_encoding
988
return self.run_bzr(*args, **kwargs)[0].decode(encoding)
990
def run_bzr_error(self, error_regexes, *args, **kwargs):
991
"""Run bzr, and check that stderr contains the supplied regexes
993
:param error_regexes: Sequence of regular expressions which
994
must each be found in the error output. The relative ordering
996
:param args: command-line arguments for bzr
997
:param kwargs: Keyword arguments which are interpreted by run_bzr
998
This function changes the default value of retcode to be 3,
999
since in most cases this is run when you expect bzr to fail.
1000
:return: (out, err) The actual output of running the command (in case you
1001
want to do more inspection)
1004
# Make sure that commit is failing because there is nothing to do
1005
self.run_bzr_error(['no changes to commit'],
1006
'commit', '-m', 'my commit comment')
1007
# Make sure --strict is handling an unknown file, rather than
1008
# giving us the 'nothing to do' error
1009
self.build_tree(['unknown'])
1010
self.run_bzr_error(['Commit refused because there are unknown files'],
1011
'commit', '--strict', '-m', 'my commit comment')
1013
kwargs.setdefault('retcode', 3)
1014
out, err = self.run_bzr(*args, **kwargs)
1015
for regex in error_regexes:
1016
self.assertContainsRe(err, regex)
1019
def run_bzr_subprocess(self, *args, **kwargs):
1020
"""Run bzr in a subprocess for testing.
1022
This starts a new Python interpreter and runs bzr in there.
1023
This should only be used for tests that have a justifiable need for
1024
this isolation: e.g. they are testing startup time, or signal
1025
handling, or early startup code, etc. Subprocess code can't be
1026
profiled or debugged so easily.
1028
:param retcode: The status code that is expected. Defaults to 0. If
1029
None is supplied, the status code is not checked.
1030
:param env_changes: A dictionary which lists changes to environment
1031
variables. A value of None will unset the env variable.
1032
The values must be strings. The change will only occur in the
1033
child, so you don't need to fix the environment after running.
1034
:param universal_newlines: Convert CRLF => LF
1035
:param allow_plugins: By default the subprocess is run with
1036
--no-plugins to ensure test reproducibility. Also, it is possible
1037
for system-wide plugins to create unexpected output on stderr,
1038
which can cause unnecessary test failures.
1040
env_changes = kwargs.get('env_changes', {})
1041
working_dir = kwargs.get('working_dir', None)
1042
allow_plugins = kwargs.get('allow_plugins', False)
1043
process = self.start_bzr_subprocess(args, env_changes=env_changes,
1044
working_dir=working_dir,
1045
allow_plugins=allow_plugins)
1046
# We distinguish between retcode=None and retcode not passed.
1047
supplied_retcode = kwargs.get('retcode', 0)
1048
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
1049
universal_newlines=kwargs.get('universal_newlines', False),
1052
def start_bzr_subprocess(self, process_args, env_changes=None,
1053
skip_if_plan_to_signal=False,
1055
allow_plugins=False):
1056
"""Start bzr in a subprocess for testing.
1058
This starts a new Python interpreter and runs bzr in there.
1059
This should only be used for tests that have a justifiable need for
1060
this isolation: e.g. they are testing startup time, or signal
1061
handling, or early startup code, etc. Subprocess code can't be
1062
profiled or debugged so easily.
1064
:param process_args: a list of arguments to pass to the bzr executable,
1065
for example `['--version']`.
1066
:param env_changes: A dictionary which lists changes to environment
1067
variables. A value of None will unset the env variable.
1068
The values must be strings. The change will only occur in the
1069
child, so you don't need to fix the environment after running.
1070
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1072
:param allow_plugins: If False (default) pass --no-plugins to bzr.
1074
:returns: Popen object for the started process.
1076
if skip_if_plan_to_signal:
1077
if not getattr(os, 'kill', None):
1078
raise TestSkipped("os.kill not available.")
1080
if env_changes is None:
1084
def cleanup_environment():
1085
for env_var, value in env_changes.iteritems():
1086
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
1088
def restore_environment():
1089
for env_var, value in old_env.iteritems():
1090
osutils.set_or_unset_env(env_var, value)
1092
bzr_path = self.get_bzr_path()
1095
if working_dir is not None:
1096
cwd = osutils.getcwd()
1097
os.chdir(working_dir)
1100
# win32 subprocess doesn't support preexec_fn
1101
# so we will avoid using it on all platforms, just to
1102
# make sure the code path is used, and we don't break on win32
1103
cleanup_environment()
1104
command = [sys.executable, bzr_path]
1105
if not allow_plugins:
1106
command.append('--no-plugins')
1107
command.extend(process_args)
1108
process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1110
restore_environment()
1116
def _popen(self, *args, **kwargs):
1117
"""Place a call to Popen.
1119
Allows tests to override this method to intercept the calls made to
1120
Popen for introspection.
1122
return Popen(*args, **kwargs)
1124
def get_bzr_path(self):
1125
"""Return the path of the 'bzr' executable for this test suite."""
1126
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1127
if not os.path.isfile(bzr_path):
1128
# We are probably installed. Assume sys.argv is the right file
1129
bzr_path = sys.argv[0]
1132
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1133
universal_newlines=False, process_args=None):
1134
"""Finish the execution of process.
1136
:param process: the Popen object returned from start_bzr_subprocess.
1137
:param retcode: The status code that is expected. Defaults to 0. If
1138
None is supplied, the status code is not checked.
1139
:param send_signal: an optional signal to send to the process.
1140
:param universal_newlines: Convert CRLF => LF
1141
:returns: (stdout, stderr)
1143
if send_signal is not None:
1144
os.kill(process.pid, send_signal)
1145
out, err = process.communicate()
1147
if universal_newlines:
1148
out = out.replace('\r\n', '\n')
1149
err = err.replace('\r\n', '\n')
1151
if retcode is not None and retcode != process.returncode:
1152
if process_args is None:
1153
process_args = "(unknown args)"
1154
mutter('Output of bzr %s:\n%s', process_args, out)
1155
mutter('Error for bzr %s:\n%s', process_args, err)
1156
self.fail('Command bzr %s failed with retcode %s != %s'
1157
% (process_args, retcode, process.returncode))
1160
def check_inventory_shape(self, inv, shape):
1161
"""Compare an inventory to a list of expected names.
1163
Fail if they are not precisely equal.
1166
shape = list(shape) # copy
1167
for path, ie in inv.entries():
1168
name = path.replace('\\', '/')
1169
if ie.kind == 'dir':
1176
self.fail("expected paths not found in inventory: %r" % shape)
1178
self.fail("unexpected paths found in inventory: %r" % extras)
1180
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
1181
a_callable=None, *args, **kwargs):
1182
"""Call callable with redirected std io pipes.
1184
Returns the return code."""
1185
if not callable(a_callable):
1186
raise ValueError("a_callable must be callable.")
1188
stdin = StringIO("")
1190
if getattr(self, "_log_file", None) is not None:
1191
stdout = self._log_file
1195
if getattr(self, "_log_file", None is not None):
1196
stderr = self._log_file
1199
real_stdin = sys.stdin
1200
real_stdout = sys.stdout
1201
real_stderr = sys.stderr
1206
return a_callable(*args, **kwargs)
1208
sys.stdout = real_stdout
1209
sys.stderr = real_stderr
1210
sys.stdin = real_stdin
1212
@symbol_versioning.deprecated_method(symbol_versioning.zero_eleven)
1213
def merge(self, branch_from, wt_to):
1214
"""A helper for tests to do a ui-less merge.
1216
This should move to the main library when someone has time to integrate
1219
# minimal ui-less merge.
1220
wt_to.branch.fetch(branch_from)
1221
base_rev = common_ancestor(branch_from.last_revision(),
1222
wt_to.branch.last_revision(),
1223
wt_to.branch.repository)
1224
merge_inner(wt_to.branch, branch_from.basis_tree(),
1225
wt_to.branch.repository.revision_tree(base_rev),
1227
wt_to.add_parent_tree_id(branch_from.last_revision())
1230
BzrTestBase = TestCase
1233
class TestCaseWithMemoryTransport(TestCase):
1234
"""Common test class for tests that do not need disk resources.
1236
Tests that need disk resources should derive from TestCaseWithTransport.
1238
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1240
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1241
a directory which does not exist. This serves to help ensure test isolation
1242
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1243
must exist. However, TestCaseWithMemoryTransport does not offer local
1244
file defaults for the transport in tests, nor does it obey the command line
1245
override, so tests that accidentally write to the common directory should
1253
def __init__(self, methodName='runTest'):
1254
# allow test parameterisation after test construction and before test
1255
# execution. Variables that the parameteriser sets need to be
1256
# ones that are not set by setUp, or setUp will trash them.
1257
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1258
self.transport_server = default_transport
1259
self.transport_readonly_server = None
1261
def failUnlessExists(self, path):
1262
"""Fail unless path, which may be abs or relative, exists."""
1263
self.failUnless(osutils.lexists(path))
1265
def failIfExists(self, path):
1266
"""Fail if path, which may be abs or relative, exists."""
1267
self.failIf(osutils.lexists(path))
1269
def get_transport(self):
1270
"""Return a writeable transport for the test scratch space"""
1271
t = get_transport(self.get_url())
1272
self.assertFalse(t.is_readonly())
1275
def get_readonly_transport(self):
1276
"""Return a readonly transport for the test scratch space
1278
This can be used to test that operations which should only need
1279
readonly access in fact do not try to write.
1281
t = get_transport(self.get_readonly_url())
1282
self.assertTrue(t.is_readonly())
1285
def create_transport_readonly_server(self):
1286
"""Create a transport server from class defined at init.
1288
This is mostly a hook for daughter classes.
1290
return self.transport_readonly_server()
1292
def get_readonly_server(self):
1293
"""Get the server instance for the readonly transport
1295
This is useful for some tests with specific servers to do diagnostics.
1297
if self.__readonly_server is None:
1298
if self.transport_readonly_server is None:
1299
# readonly decorator requested
1300
# bring up the server
1302
self.__readonly_server = ReadonlyServer()
1303
self.__readonly_server.setUp(self.__server)
1305
self.__readonly_server = self.create_transport_readonly_server()
1306
self.__readonly_server.setUp()
1307
self.addCleanup(self.__readonly_server.tearDown)
1308
return self.__readonly_server
1310
def get_readonly_url(self, relpath=None):
1311
"""Get a URL for the readonly transport.
1313
This will either be backed by '.' or a decorator to the transport
1314
used by self.get_url()
1315
relpath provides for clients to get a path relative to the base url.
1316
These should only be downwards relative, not upwards.
1318
base = self.get_readonly_server().get_url()
1319
if relpath is not None:
1320
if not base.endswith('/'):
1322
base = base + relpath
1325
def get_server(self):
1326
"""Get the read/write server instance.
1328
This is useful for some tests with specific servers that need
1331
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1332
is no means to override it.
1334
if self.__server is None:
1335
self.__server = MemoryServer()
1336
self.__server.setUp()
1337
self.addCleanup(self.__server.tearDown)
1338
return self.__server
1340
def get_url(self, relpath=None):
1341
"""Get a URL (or maybe a path) for the readwrite transport.
1343
This will either be backed by '.' or to an equivalent non-file based
1345
relpath provides for clients to get a path relative to the base url.
1346
These should only be downwards relative, not upwards.
1348
base = self.get_server().get_url()
1349
if relpath is not None and relpath != '.':
1350
if not base.endswith('/'):
1352
# XXX: Really base should be a url; we did after all call
1353
# get_url()! But sometimes it's just a path (from
1354
# LocalAbspathServer), and it'd be wrong to append urlescaped data
1355
# to a non-escaped local path.
1356
if base.startswith('./') or base.startswith('/'):
1359
base += urlutils.escape(relpath)
1362
def _make_test_root(self):
1363
if TestCaseWithMemoryTransport.TEST_ROOT is not None:
1367
root = u'test%04d.tmp' % i
1371
if e.errno == errno.EEXIST:
1376
# successfully created
1377
TestCaseWithMemoryTransport.TEST_ROOT = osutils.abspath(root)
1379
# make a fake bzr directory there to prevent any tests propagating
1380
# up onto the source directory's real branch
1381
bzrdir.BzrDir.create_standalone_workingtree(
1382
TestCaseWithMemoryTransport.TEST_ROOT)
1384
def makeAndChdirToTestDir(self):
1385
"""Create a temporary directories for this one test.
1387
This must set self.test_home_dir and self.test_dir and chdir to
1390
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1392
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1393
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1394
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1396
def make_branch(self, relpath, format=None):
1397
"""Create a branch on the transport at relpath."""
1398
repo = self.make_repository(relpath, format=format)
1399
return repo.bzrdir.create_branch()
1401
def make_bzrdir(self, relpath, format=None):
1403
# might be a relative or absolute path
1404
maybe_a_url = self.get_url(relpath)
1405
segments = maybe_a_url.rsplit('/', 1)
1406
t = get_transport(maybe_a_url)
1407
if len(segments) > 1 and segments[-1] not in ('', '.'):
1410
except errors.FileExists:
1413
format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
1414
return format.initialize_on_transport(t)
1415
except errors.UninitializableFormat:
1416
raise TestSkipped("Format %s is not initializable." % format)
1418
def make_repository(self, relpath, shared=False, format=None):
1419
"""Create a repository on our default transport at relpath."""
1420
made_control = self.make_bzrdir(relpath, format=format)
1421
return made_control.create_repository(shared=shared)
1423
def make_branch_and_memory_tree(self, relpath, format=None):
1424
"""Create a branch on the default transport and a MemoryTree for it."""
1425
b = self.make_branch(relpath, format=format)
1426
return memorytree.MemoryTree.create_on_branch(b)
1428
def overrideEnvironmentForTesting(self):
1429
os.environ['HOME'] = self.test_home_dir
1430
os.environ['APPDATA'] = self.test_home_dir
1433
super(TestCaseWithMemoryTransport, self).setUp()
1434
self._make_test_root()
1435
_currentdir = os.getcwdu()
1436
def _leaveDirectory():
1437
os.chdir(_currentdir)
1438
self.addCleanup(_leaveDirectory)
1439
self.makeAndChdirToTestDir()
1440
self.overrideEnvironmentForTesting()
1441
self.__readonly_server = None
1442
self.__server = None
1445
class TestCaseInTempDir(TestCaseWithMemoryTransport):
1446
"""Derived class that runs a test within a temporary directory.
1448
This is useful for tests that need to create a branch, etc.
1450
The directory is created in a slightly complex way: for each
1451
Python invocation, a new temporary top-level directory is created.
1452
All test cases create their own directory within that. If the
1453
tests complete successfully, the directory is removed.
1455
InTempDir is an old alias for FunctionalTestCase.
1458
OVERRIDE_PYTHON = 'python'
1460
def check_file_contents(self, filename, expect):
1461
self.log("check contents of file %s" % filename)
1462
contents = file(filename, 'r').read()
1463
if contents != expect:
1464
self.log("expected: %r" % expect)
1465
self.log("actually: %r" % contents)
1466
self.fail("contents of %s not as expected" % filename)
1468
def makeAndChdirToTestDir(self):
1469
"""See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
1471
For TestCaseInTempDir we create a temporary directory based on the test
1472
name and then create two subdirs - test and home under it.
1474
# shorten the name, to avoid test failures due to path length
1475
short_id = self.id().replace('bzrlib.tests.', '') \
1476
.replace('__main__.', '')[-100:]
1477
# it's possible the same test class is run several times for
1478
# parameterized tests, so make sure the names don't collide.
1482
candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
1484
candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
1485
if os.path.exists(candidate_dir):
1489
os.mkdir(candidate_dir)
1490
self.test_home_dir = candidate_dir + '/home'
1491
os.mkdir(self.test_home_dir)
1492
self.test_dir = candidate_dir + '/work'
1493
os.mkdir(self.test_dir)
1494
os.chdir(self.test_dir)
1497
def build_tree(self, shape, line_endings='native', transport=None):
1498
"""Build a test tree according to a pattern.
1500
shape is a sequence of file specifications. If the final
1501
character is '/', a directory is created.
1503
This assumes that all the elements in the tree being built are new.
1505
This doesn't add anything to a branch.
1506
:param line_endings: Either 'binary' or 'native'
1507
in binary mode, exact contents are written
1508
in native mode, the line endings match the
1509
default platform endings.
1511
:param transport: A transport to write to, for building trees on
1512
VFS's. If the transport is readonly or None,
1513
"." is opened automatically.
1515
# It's OK to just create them using forward slashes on windows.
1516
if transport is None or transport.is_readonly():
1517
transport = get_transport(".")
1519
self.assert_(isinstance(name, basestring))
1521
transport.mkdir(urlutils.escape(name[:-1]))
1523
if line_endings == 'binary':
1525
elif line_endings == 'native':
1528
raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
1529
content = "contents of %s%s" % (name.encode('utf-8'), end)
1530
# Technically 'put()' is the right command. However, put
1531
# uses an AtomicFile, which requires an extra rename into place
1532
# As long as the files didn't exist in the past, append() will
1533
# do the same thing as put()
1534
# On jam's machine, make_kernel_like_tree is:
1535
# put: 4.5-7.5s (averaging 6s)
1537
# put_non_atomic: 2.9-4.5s
1538
transport.put_bytes_non_atomic(urlutils.escape(name), content)
1540
def build_tree_contents(self, shape):
1541
build_tree_contents(shape)
1543
def assertFileEqual(self, content, path):
1544
"""Fail if path does not contain 'content'."""
1545
self.failUnless(osutils.lexists(path))
1546
# TODO: jam 20060427 Shouldn't this be 'rb'?
1547
self.assertEqualDiff(content, open(path, 'r').read())
1550
class TestCaseWithTransport(TestCaseInTempDir):
1551
"""A test case that provides get_url and get_readonly_url facilities.
1553
These back onto two transport servers, one for readonly access and one for
1556
If no explicit class is provided for readonly access, a
1557
ReadonlyTransportDecorator is used instead which allows the use of non disk
1558
based read write transports.
1560
If an explicit class is provided for readonly access, that server and the
1561
readwrite one must both define get_url() as resolving to os.getcwd().
1564
def create_transport_server(self):
1565
"""Create a transport server from class defined at init.
1567
This is mostly a hook for daughter classes.
1569
return self.transport_server()
1571
def get_server(self):
1572
"""See TestCaseWithMemoryTransport.
1574
This is useful for some tests with specific servers that need
1577
if self.__server is None:
1578
self.__server = self.create_transport_server()
1579
self.__server.setUp()
1580
self.addCleanup(self.__server.tearDown)
1581
return self.__server
1583
def make_branch_and_tree(self, relpath, format=None):
1584
"""Create a branch on the transport and a tree locally.
1586
If the transport is not a LocalTransport, the Tree can't be created on
1587
the transport. In that case the working tree is created in the local
1588
directory, and the returned tree's branch and repository will also be
1591
This will fail if the original default transport for this test
1592
case wasn't backed by the working directory, as the branch won't
1593
be on disk for us to open it.
1595
:param format: The BzrDirFormat.
1596
:returns: the WorkingTree.
1598
# TODO: always use the local disk path for the working tree,
1599
# this obviously requires a format that supports branch references
1600
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
1602
b = self.make_branch(relpath, format=format)
1604
return b.bzrdir.create_workingtree()
1605
except errors.NotLocalUrl:
1606
# We can only make working trees locally at the moment. If the
1607
# transport can't support them, then reopen the branch on a local
1608
# transport, and create the working tree there.
1610
# Possibly we should instead keep
1611
# the non-disk-backed branch and create a local checkout?
1612
bd = bzrdir.BzrDir.open(relpath)
1613
return bd.create_workingtree()
1615
def assertIsDirectory(self, relpath, transport):
1616
"""Assert that relpath within transport is a directory.
1618
This may not be possible on all transports; in that case it propagates
1619
a TransportNotPossible.
1622
mode = transport.stat(relpath).st_mode
1623
except errors.NoSuchFile:
1624
self.fail("path %s is not a directory; no such file"
1626
if not stat.S_ISDIR(mode):
1627
self.fail("path %s is not a directory; has mode %#o"
1631
super(TestCaseWithTransport, self).setUp()
1632
self.__server = None
1635
class ChrootedTestCase(TestCaseWithTransport):
1636
"""A support class that provides readonly urls outside the local namespace.
1638
This is done by checking if self.transport_server is a MemoryServer. if it
1639
is then we are chrooted already, if it is not then an HttpServer is used
1642
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
1643
be used without needed to redo it when a different
1644
subclass is in use ?
1648
super(ChrootedTestCase, self).setUp()
1649
if not self.transport_server == MemoryServer:
1650
self.transport_readonly_server = HttpServer
1653
def filter_suite_by_re(suite, pattern):
1654
result = TestUtil.TestSuite()
1655
filter_re = re.compile(pattern)
1656
for test in iter_suite_tests(suite):
1657
if filter_re.search(test.id()):
1658
result.addTest(test)
1662
def run_suite(suite, name='test', verbose=False, pattern=".*",
1663
stop_on_failure=False, keep_output=False,
1664
transport=None, lsprof_timed=None, bench_history=None):
1665
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
1670
runner = TextTestRunner(stream=sys.stdout,
1672
verbosity=verbosity,
1673
keep_output=keep_output,
1674
bench_history=bench_history)
1675
runner.stop_on_failure=stop_on_failure
1677
suite = filter_suite_by_re(suite, pattern)
1678
result = runner.run(suite)
1679
return result.wasSuccessful()
1682
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
1685
test_suite_factory=None,
1687
bench_history=None):
1688
"""Run the whole test suite under the enhanced runner"""
1689
# XXX: Very ugly way to do this...
1690
# Disable warning about old formats because we don't want it to disturb
1691
# any blackbox tests.
1692
from bzrlib import repository
1693
repository._deprecation_warning_done = True
1695
global default_transport
1696
if transport is None:
1697
transport = default_transport
1698
old_transport = default_transport
1699
default_transport = transport
1701
if test_suite_factory is None:
1702
suite = test_suite()
1704
suite = test_suite_factory()
1705
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
1706
stop_on_failure=stop_on_failure, keep_output=keep_output,
1707
transport=transport,
1708
lsprof_timed=lsprof_timed,
1709
bench_history=bench_history)
1711
default_transport = old_transport
1715
"""Build and return TestSuite for the whole of bzrlib.
1717
This function can be replaced if you need to change the default test
1718
suite on a global basis, but it is not encouraged.
1721
'bzrlib.tests.test_ancestry',
1722
'bzrlib.tests.test_annotate',
1723
'bzrlib.tests.test_api',
1724
'bzrlib.tests.test_atomicfile',
1725
'bzrlib.tests.test_bad_files',
1726
'bzrlib.tests.test_branch',
1727
'bzrlib.tests.test_bundle',
1728
'bzrlib.tests.test_bzrdir',
1729
'bzrlib.tests.test_cache_utf8',
1730
'bzrlib.tests.test_commands',
1731
'bzrlib.tests.test_commit',
1732
'bzrlib.tests.test_commit_merge',
1733
'bzrlib.tests.test_config',
1734
'bzrlib.tests.test_conflicts',
1735
'bzrlib.tests.test_decorators',
1736
'bzrlib.tests.test_diff',
1737
'bzrlib.tests.test_doc_generate',
1738
'bzrlib.tests.test_errors',
1739
'bzrlib.tests.test_escaped_store',
1740
'bzrlib.tests.test_fetch',
1741
'bzrlib.tests.test_ftp_transport',
1742
'bzrlib.tests.test_generate_ids',
1743
'bzrlib.tests.test_globbing',
1744
'bzrlib.tests.test_gpg',
1745
'bzrlib.tests.test_graph',
1746
'bzrlib.tests.test_hashcache',
1747
'bzrlib.tests.test_http',
1748
'bzrlib.tests.test_http_response',
1749
'bzrlib.tests.test_identitymap',
1750
'bzrlib.tests.test_ignores',
1751
'bzrlib.tests.test_inv',
1752
'bzrlib.tests.test_knit',
1753
'bzrlib.tests.test_lazy_import',
1754
'bzrlib.tests.test_lazy_regex',
1755
'bzrlib.tests.test_lockdir',
1756
'bzrlib.tests.test_lockable_files',
1757
'bzrlib.tests.test_log',
1758
'bzrlib.tests.test_memorytree',
1759
'bzrlib.tests.test_merge',
1760
'bzrlib.tests.test_merge3',
1761
'bzrlib.tests.test_merge_core',
1762
'bzrlib.tests.test_missing',
1763
'bzrlib.tests.test_msgeditor',
1764
'bzrlib.tests.test_nonascii',
1765
'bzrlib.tests.test_options',
1766
'bzrlib.tests.test_osutils',
1767
'bzrlib.tests.test_patch',
1768
'bzrlib.tests.test_patches',
1769
'bzrlib.tests.test_permissions',
1770
'bzrlib.tests.test_plugins',
1771
'bzrlib.tests.test_progress',
1772
'bzrlib.tests.test_reconcile',
1773
'bzrlib.tests.test_registry',
1774
'bzrlib.tests.test_repository',
1775
'bzrlib.tests.test_revert',
1776
'bzrlib.tests.test_revision',
1777
'bzrlib.tests.test_revisionnamespaces',
1778
'bzrlib.tests.test_revisiontree',
1779
'bzrlib.tests.test_rio',
1780
'bzrlib.tests.test_sampler',
1781
'bzrlib.tests.test_selftest',
1782
'bzrlib.tests.test_setup',
1783
'bzrlib.tests.test_sftp_transport',
1784
'bzrlib.tests.test_smart_add',
1785
'bzrlib.tests.test_smart_transport',
1786
'bzrlib.tests.test_source',
1787
'bzrlib.tests.test_status',
1788
'bzrlib.tests.test_store',
1789
'bzrlib.tests.test_symbol_versioning',
1790
'bzrlib.tests.test_testament',
1791
'bzrlib.tests.test_textfile',
1792
'bzrlib.tests.test_textmerge',
1793
'bzrlib.tests.test_trace',
1794
'bzrlib.tests.test_transactions',
1795
'bzrlib.tests.test_transform',
1796
'bzrlib.tests.test_transport',
1797
'bzrlib.tests.test_tree',
1798
'bzrlib.tests.test_treebuilder',
1799
'bzrlib.tests.test_tsort',
1800
'bzrlib.tests.test_tuned_gzip',
1801
'bzrlib.tests.test_ui',
1802
'bzrlib.tests.test_upgrade',
1803
'bzrlib.tests.test_urlutils',
1804
'bzrlib.tests.test_versionedfile',
1805
'bzrlib.tests.test_version',
1806
'bzrlib.tests.test_version_info',
1807
'bzrlib.tests.test_weave',
1808
'bzrlib.tests.test_whitebox',
1809
'bzrlib.tests.test_workingtree',
1810
'bzrlib.tests.test_wsgi',
1811
'bzrlib.tests.test_xml',
1813
test_transport_implementations = [
1814
'bzrlib.tests.test_transport_implementations',
1815
'bzrlib.tests.test_read_bundle',
1817
suite = TestUtil.TestSuite()
1818
loader = TestUtil.TestLoader()
1819
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1820
from bzrlib.transport import TransportTestProviderAdapter
1821
adapter = TransportTestProviderAdapter()
1822
adapt_modules(test_transport_implementations, adapter, loader, suite)
1823
for package in packages_to_test():
1824
suite.addTest(package.test_suite())
1825
for m in MODULES_TO_TEST:
1826
suite.addTest(loader.loadTestsFromModule(m))
1827
for m in MODULES_TO_DOCTEST:
1829
suite.addTest(doctest.DocTestSuite(m))
1830
except ValueError, e:
1831
print '**failed to get doctest for: %s\n%s' %(m,e)
1833
for name, plugin in bzrlib.plugin.all_plugins().items():
1834
if getattr(plugin, 'test_suite', None) is not None:
1835
suite.addTest(plugin.test_suite())
1839
def adapt_modules(mods_list, adapter, loader, suite):
1840
"""Adapt the modules in mods_list using adapter and add to suite."""
1841
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1842
suite.addTests(adapter.adapt(test))
1845
def clean_selftest_output(root=None, quiet=False):
1846
"""Remove all selftest output directories from root directory.
1848
:param root: root directory for clean
1849
(if ommitted or None then clean current directory).
1850
:param quiet: suppress report about deleting directories
1855
re_dir = re.compile(r'''test\d\d\d\d\.tmp''')
1858
for i in os.listdir(root):
1859
if os.path.isdir(i) and re_dir.match(i):
1861
print 'delete directory:', i
added
removed
bzrlib/tests/__init__.py
