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, 60))
382
def report_error(self, test, err):
383
self.error_count += 1
384
self.stream.writeln('ERROR %s\n %s'
385
% (self._testTimeString(), err[1]))
387
def report_failure(self, test, err):
388
self.failure_count += 1
389
self.stream.writeln('FAIL %s\n %s'
390
% (self._testTimeString(), err[1]))
392
def report_success(self, test):
393
self.stream.writeln(' OK %s' % self._testTimeString())
394
for bench_called, stats in getattr(test, '_benchcalls', []):
395
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
396
stats.pprint(file=self.stream)
399
def report_skip(self, test, skip_excinfo):
400
print >>self.stream, ' SKIP %s' % self._testTimeString()
401
print >>self.stream, ' %s' % skip_excinfo[1]
404
class TextTestRunner(object):
405
stop_on_failure = False
413
self.stream = unittest._WritelnDecorator(stream)
414
self.descriptions = descriptions
415
self.verbosity = verbosity
416
self.keep_output = keep_output
417
self._bench_history = bench_history
420
"Run the given test case or test suite."
421
startTime = time.time()
422
if self.verbosity == 1:
423
result_class = TextTestResult
424
elif self.verbosity >= 2:
425
result_class = VerboseTestResult
426
result = result_class(self.stream,
429
bench_history=self._bench_history,
430
num_tests=test.countTestCases(),
432
result.stop_early = self.stop_on_failure
433
result.report_starting()
435
stopTime = time.time()
436
timeTaken = stopTime - startTime
438
self.stream.writeln(result.separator2)
439
run = result.testsRun
440
self.stream.writeln("Ran %d test%s in %.3fs" %
441
(run, run != 1 and "s" or "", timeTaken))
442
self.stream.writeln()
443
if not result.wasSuccessful():
444
self.stream.write("FAILED (")
445
failed, errored = map(len, (result.failures, result.errors))
447
self.stream.write("failures=%d" % failed)
449
if failed: self.stream.write(", ")
450
self.stream.write("errors=%d" % errored)
451
self.stream.writeln(")")
453
self.stream.writeln("OK")
454
result.report_cleaning_up()
455
# This is still a little bogus,
456
# but only a little. Folk not using our testrunner will
457
# have to delete their temp directories themselves.
458
test_root = TestCaseWithMemoryTransport.TEST_ROOT
459
if result.wasSuccessful() or not self.keep_output:
460
if test_root is not None:
461
# If LANG=C we probably have created some bogus paths
462
# which rmtree(unicode) will fail to delete
463
# so make sure we are using rmtree(str) to delete everything
464
# except on win32, where rmtree(str) will fail
465
# since it doesn't have the property of byte-stream paths
466
# (they are either ascii or mbcs)
467
if sys.platform == 'win32':
468
# make sure we are using the unicode win32 api
469
test_root = unicode(test_root)
471
test_root = test_root.encode(
472
sys.getfilesystemencoding())
473
osutils.rmtree(test_root)
475
note("Failed tests working directories are in '%s'\n", test_root)
476
TestCaseWithMemoryTransport.TEST_ROOT = None
481
def iter_suite_tests(suite):
482
"""Return all tests in a suite, recursing through nested suites"""
483
for item in suite._tests:
484
if isinstance(item, unittest.TestCase):
486
elif isinstance(item, unittest.TestSuite):
487
for r in iter_suite_tests(item):
490
raise Exception('unknown object %r inside test suite %r'
494
class TestSkipped(Exception):
495
"""Indicates that a test was intentionally skipped, rather than failing."""
498
class CommandFailed(Exception):
502
class StringIOWrapper(object):
503
"""A wrapper around cStringIO which just adds an encoding attribute.
505
Internally we can check sys.stdout to see what the output encoding
506
should be. However, cStringIO has no encoding attribute that we can
507
set. So we wrap it instead.
512
def __init__(self, s=None):
514
self.__dict__['_cstring'] = StringIO(s)
516
self.__dict__['_cstring'] = StringIO()
518
def __getattr__(self, name, getattr=getattr):
519
return getattr(self.__dict__['_cstring'], name)
521
def __setattr__(self, name, val):
522
if name == 'encoding':
523
self.__dict__['encoding'] = val
525
return setattr(self._cstring, name, val)
528
class TestCase(unittest.TestCase):
529
"""Base class for bzr unit tests.
531
Tests that need access to disk resources should subclass
532
TestCaseInTempDir not TestCase.
534
Error and debug log messages are redirected from their usual
535
location into a temporary file, the contents of which can be
536
retrieved by _get_log(). We use a real OS file, not an in-memory object,
537
so that it can also capture file IO. When the test completes this file
538
is read into memory and removed from disk.
540
There are also convenience functions to invoke bzr's command-line
541
routine, and to build and check bzr trees.
543
In addition to the usual method of overriding tearDown(), this class also
544
allows subclasses to register functions into the _cleanups list, which is
545
run in order as the object is torn down. It's less likely this will be
546
accidentally overlooked.
549
_log_file_name = None
551
_keep_log_file = False
552
# record lsprof data when performing benchmark calls.
553
_gather_lsprof_in_benchmarks = False
555
def __init__(self, methodName='testMethod'):
556
super(TestCase, self).__init__(methodName)
560
unittest.TestCase.setUp(self)
561
self._cleanEnvironment()
562
bzrlib.trace.disable_default_logging()
565
self._benchcalls = []
566
self._benchtime = None
568
def _silenceUI(self):
569
"""Turn off UI for duration of test"""
570
# by default the UI is off; tests can turn it on if they want it.
571
saved = bzrlib.ui.ui_factory
573
bzrlib.ui.ui_factory = saved
574
bzrlib.ui.ui_factory = bzrlib.ui.SilentUIFactory()
575
self.addCleanup(_restore)
577
def _ndiff_strings(self, a, b):
578
"""Return ndiff between two strings containing lines.
580
A trailing newline is added if missing to make the strings
582
if b and b[-1] != '\n':
584
if a and a[-1] != '\n':
586
difflines = difflib.ndiff(a.splitlines(True),
588
linejunk=lambda x: False,
589
charjunk=lambda x: False)
590
return ''.join(difflines)
592
def assertEqualDiff(self, a, b, message=None):
593
"""Assert two texts are equal, if not raise an exception.
595
This is intended for use with multi-line strings where it can
596
be hard to find the differences by eye.
598
# TODO: perhaps override assertEquals to call this for strings?
602
message = "texts not equal:\n"
603
raise AssertionError(message +
604
self._ndiff_strings(a, b))
606
def assertEqualMode(self, mode, mode_test):
607
self.assertEqual(mode, mode_test,
608
'mode mismatch %o != %o' % (mode, mode_test))
610
def assertStartsWith(self, s, prefix):
611
if not s.startswith(prefix):
612
raise AssertionError('string %r does not start with %r' % (s, prefix))
614
def assertEndsWith(self, s, suffix):
615
"""Asserts that s ends with suffix."""
616
if not s.endswith(suffix):
617
raise AssertionError('string %r does not end with %r' % (s, suffix))
619
def assertContainsRe(self, haystack, needle_re):
620
"""Assert that a contains something matching a regular expression."""
621
if not re.search(needle_re, haystack):
622
raise AssertionError('pattern "%s" not found in "%s"'
623
% (needle_re, haystack))
625
def assertNotContainsRe(self, haystack, needle_re):
626
"""Assert that a does not match a regular expression"""
627
if re.search(needle_re, haystack):
628
raise AssertionError('pattern "%s" found in "%s"'
629
% (needle_re, haystack))
631
def assertSubset(self, sublist, superlist):
632
"""Assert that every entry in sublist is present in superlist."""
634
for entry in sublist:
635
if entry not in superlist:
636
missing.append(entry)
638
raise AssertionError("value(s) %r not present in container %r" %
639
(missing, superlist))
641
def assertListRaises(self, excClass, func, *args, **kwargs):
642
"""Fail unless excClass is raised when the iterator from func is used.
644
Many functions can return generators this makes sure
645
to wrap them in a list() call to make sure the whole generator
646
is run, and that the proper exception is raised.
649
list(func(*args, **kwargs))
653
if getattr(excClass,'__name__', None) is not None:
654
excName = excClass.__name__
656
excName = str(excClass)
657
raise self.failureException, "%s not raised" % excName
659
def assertIs(self, left, right):
660
if not (left is right):
661
raise AssertionError("%r is not %r." % (left, right))
663
def assertTransportMode(self, transport, path, mode):
664
"""Fail if a path does not have mode mode.
666
If modes are not supported on this transport, the assertion is ignored.
668
if not transport._can_roundtrip_unix_modebits():
670
path_stat = transport.stat(path)
671
actual_mode = stat.S_IMODE(path_stat.st_mode)
672
self.assertEqual(mode, actual_mode,
673
'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
675
def assertIsInstance(self, obj, kls):
676
"""Fail if obj is not an instance of kls"""
677
if not isinstance(obj, kls):
678
self.fail("%r is an instance of %s rather than %s" % (
679
obj, obj.__class__, kls))
681
def _capture_warnings(self, a_callable, *args, **kwargs):
682
"""A helper for callDeprecated and applyDeprecated.
684
:param a_callable: A callable to call.
685
:param args: The positional arguments for the callable
686
:param kwargs: The keyword arguments for the callable
687
:return: A tuple (warnings, result). result is the result of calling
688
a_callable(*args, **kwargs).
691
def capture_warnings(msg, cls=None, stacklevel=None):
692
# we've hooked into a deprecation specific callpath,
693
# only deprecations should getting sent via it.
694
self.assertEqual(cls, DeprecationWarning)
695
local_warnings.append(msg)
696
original_warning_method = symbol_versioning.warn
697
symbol_versioning.set_warning_method(capture_warnings)
699
result = a_callable(*args, **kwargs)
701
symbol_versioning.set_warning_method(original_warning_method)
702
return (local_warnings, result)
704
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
705
"""Call a deprecated callable without warning the user.
707
:param deprecation_format: The deprecation format that the callable
708
should have been deprecated with. This is the same type as the
709
parameter to deprecated_method/deprecated_function. If the
710
callable is not deprecated with this format, an assertion error
712
:param a_callable: A callable to call. This may be a bound method or
713
a regular function. It will be called with *args and **kwargs.
714
:param args: The positional arguments for the callable
715
:param kwargs: The keyword arguments for the callable
716
:return: The result of a_callable(*args, **kwargs)
718
call_warnings, result = self._capture_warnings(a_callable,
720
expected_first_warning = symbol_versioning.deprecation_string(
721
a_callable, deprecation_format)
722
if len(call_warnings) == 0:
723
self.fail("No assertion generated by call to %s" %
725
self.assertEqual(expected_first_warning, call_warnings[0])
728
def callDeprecated(self, expected, callable, *args, **kwargs):
729
"""Assert that a callable is deprecated in a particular way.
731
This is a very precise test for unusual requirements. The
732
applyDeprecated helper function is probably more suited for most tests
733
as it allows you to simply specify the deprecation format being used
734
and will ensure that that is issued for the function being called.
736
:param expected: a list of the deprecation warnings expected, in order
737
:param callable: The callable to call
738
:param args: The positional arguments for the callable
739
:param kwargs: The keyword arguments for the callable
741
call_warnings, result = self._capture_warnings(callable,
743
self.assertEqual(expected, call_warnings)
746
def _startLogFile(self):
747
"""Send bzr and test log messages to a temporary file.
749
The file is removed as the test is torn down.
751
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
752
self._log_file = os.fdopen(fileno, 'w+')
753
self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
754
self._log_file_name = name
755
self.addCleanup(self._finishLogFile)
757
def _finishLogFile(self):
758
"""Finished with the log file.
760
Close the file and delete it, unless setKeepLogfile was called.
762
if self._log_file is None:
764
bzrlib.trace.disable_test_log(self._log_nonce)
765
self._log_file.close()
766
self._log_file = None
767
if not self._keep_log_file:
768
os.remove(self._log_file_name)
769
self._log_file_name = None
771
def setKeepLogfile(self):
772
"""Make the logfile not be deleted when _finishLogFile is called."""
773
self._keep_log_file = True
775
def addCleanup(self, callable):
776
"""Arrange to run a callable when this case is torn down.
778
Callables are run in the reverse of the order they are registered,
779
ie last-in first-out.
781
if callable in self._cleanups:
782
raise ValueError("cleanup function %r already registered on %s"
784
self._cleanups.append(callable)
786
def _cleanEnvironment(self):
788
'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
790
'APPDATA': os.getcwd(),
792
'BZREMAIL': None, # may still be present in the environment
794
'BZR_PROGRESS_BAR': None,
797
self.addCleanup(self._restoreEnvironment)
798
for name, value in new_env.iteritems():
799
self._captureVar(name, value)
801
def _captureVar(self, name, newvalue):
802
"""Set an environment variable, and reset it when finished."""
803
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
805
def _restoreEnvironment(self):
806
for name, value in self.__old_env.iteritems():
807
osutils.set_or_unset_env(name, value)
811
unittest.TestCase.tearDown(self)
813
def time(self, callable, *args, **kwargs):
814
"""Run callable and accrue the time it takes to the benchmark time.
816
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
817
this will cause lsprofile statistics to be gathered and stored in
820
if self._benchtime is None:
824
if not self._gather_lsprof_in_benchmarks:
825
return callable(*args, **kwargs)
827
# record this benchmark
828
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
830
self._benchcalls.append(((callable, args, kwargs), stats))
833
self._benchtime += time.time() - start
835
def _runCleanups(self):
836
"""Run registered cleanup functions.
838
This should only be called from TestCase.tearDown.
840
# TODO: Perhaps this should keep running cleanups even if
842
for cleanup_fn in reversed(self._cleanups):
845
def log(self, *args):
848
def _get_log(self, keep_log_file=False):
849
"""Return as a string the log for this test. If the file is still
850
on disk and keep_log_file=False, delete the log file and store the
851
content in self._log_contents."""
852
# flush the log file, to get all content
854
bzrlib.trace._trace_file.flush()
855
if self._log_contents:
856
return self._log_contents
857
if self._log_file_name is not None:
858
logfile = open(self._log_file_name)
860
log_contents = logfile.read()
863
if not keep_log_file:
864
self._log_contents = log_contents
865
os.remove(self._log_file_name)
868
return "DELETED log file to reduce memory footprint"
870
def capture(self, cmd, retcode=0):
871
"""Shortcut that splits cmd into words, runs, and returns stdout"""
872
return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
874
def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None,
876
"""Invoke bzr and return (stdout, stderr).
878
Useful for code that wants to check the contents of the
879
output, the way error messages are presented, etc.
881
This should be the main method for tests that want to exercise the
882
overall behavior of the bzr application (rather than a unit test
883
or a functional test of the library.)
885
Much of the old code runs bzr by forking a new copy of Python, but
886
that is slower, harder to debug, and generally not necessary.
888
This runs bzr through the interface that catches and reports
889
errors, and with logging set to something approximating the
890
default, so that error reporting can be checked.
892
:param argv: arguments to invoke bzr
893
:param retcode: expected return code, or None for don't-care.
894
:param encoding: encoding for sys.stdout and sys.stderr
895
:param stdin: A string to be used as stdin for the command.
896
:param working_dir: Change to this directory before running
899
encoding = bzrlib.user_encoding
900
if stdin is not None:
901
stdin = StringIO(stdin)
902
stdout = StringIOWrapper()
903
stderr = StringIOWrapper()
904
stdout.encoding = encoding
905
stderr.encoding = encoding
907
self.log('run bzr: %r', argv)
908
# FIXME: don't call into logging here
909
handler = logging.StreamHandler(stderr)
910
handler.setLevel(logging.INFO)
911
logger = logging.getLogger('')
912
logger.addHandler(handler)
913
old_ui_factory = bzrlib.ui.ui_factory
914
bzrlib.ui.ui_factory = bzrlib.tests.blackbox.TestUIFactory(
917
bzrlib.ui.ui_factory.stdin = stdin
920
if working_dir is not None:
921
cwd = osutils.getcwd()
922
os.chdir(working_dir)
925
saved_debug_flags = frozenset(debug.debug_flags)
926
debug.debug_flags.clear()
928
result = self.apply_redirected(stdin, stdout, stderr,
929
bzrlib.commands.run_bzr_catch_errors,
932
debug.debug_flags.update(saved_debug_flags)
934
logger.removeHandler(handler)
935
bzrlib.ui.ui_factory = old_ui_factory
939
out = stdout.getvalue()
940
err = stderr.getvalue()
942
self.log('output:\n%r', out)
944
self.log('errors:\n%r', err)
945
if retcode is not None:
946
self.assertEquals(retcode, result)
949
def run_bzr(self, *args, **kwargs):
950
"""Invoke bzr, as if it were run from the command line.
952
This should be the main method for tests that want to exercise the
953
overall behavior of the bzr application (rather than a unit test
954
or a functional test of the library.)
956
This sends the stdout/stderr results into the test's log,
957
where it may be useful for debugging. See also run_captured.
959
:param stdin: A string to be used as stdin for the command.
961
retcode = kwargs.pop('retcode', 0)
962
encoding = kwargs.pop('encoding', None)
963
stdin = kwargs.pop('stdin', None)
964
working_dir = kwargs.pop('working_dir', None)
965
return self.run_bzr_captured(args, retcode=retcode, encoding=encoding,
966
stdin=stdin, working_dir=working_dir)
968
def run_bzr_decode(self, *args, **kwargs):
969
if 'encoding' in kwargs:
970
encoding = kwargs['encoding']
972
encoding = bzrlib.user_encoding
973
return self.run_bzr(*args, **kwargs)[0].decode(encoding)
975
def run_bzr_error(self, error_regexes, *args, **kwargs):
976
"""Run bzr, and check that stderr contains the supplied regexes
978
:param error_regexes: Sequence of regular expressions which
979
must each be found in the error output. The relative ordering
981
:param args: command-line arguments for bzr
982
:param kwargs: Keyword arguments which are interpreted by run_bzr
983
This function changes the default value of retcode to be 3,
984
since in most cases this is run when you expect bzr to fail.
985
:return: (out, err) The actual output of running the command (in case you
986
want to do more inspection)
989
# Make sure that commit is failing because there is nothing to do
990
self.run_bzr_error(['no changes to commit'],
991
'commit', '-m', 'my commit comment')
992
# Make sure --strict is handling an unknown file, rather than
993
# giving us the 'nothing to do' error
994
self.build_tree(['unknown'])
995
self.run_bzr_error(['Commit refused because there are unknown files'],
996
'commit', '--strict', '-m', 'my commit comment')
998
kwargs.setdefault('retcode', 3)
999
out, err = self.run_bzr(*args, **kwargs)
1000
for regex in error_regexes:
1001
self.assertContainsRe(err, regex)
1004
def run_bzr_subprocess(self, *args, **kwargs):
1005
"""Run bzr in a subprocess for testing.
1007
This starts a new Python interpreter and runs bzr in there.
1008
This should only be used for tests that have a justifiable need for
1009
this isolation: e.g. they are testing startup time, or signal
1010
handling, or early startup code, etc. Subprocess code can't be
1011
profiled or debugged so easily.
1013
:param retcode: The status code that is expected. Defaults to 0. If
1014
None is supplied, the status code is not checked.
1015
:param env_changes: A dictionary which lists changes to environment
1016
variables. A value of None will unset the env variable.
1017
The values must be strings. The change will only occur in the
1018
child, so you don't need to fix the environment after running.
1019
:param universal_newlines: Convert CRLF => LF
1020
:param allow_plugins: By default the subprocess is run with
1021
--no-plugins to ensure test reproducibility. Also, it is possible
1022
for system-wide plugins to create unexpected output on stderr,
1023
which can cause unnecessary test failures.
1025
env_changes = kwargs.get('env_changes', {})
1026
working_dir = kwargs.get('working_dir', None)
1027
allow_plugins = kwargs.get('allow_plugins', False)
1028
process = self.start_bzr_subprocess(args, env_changes=env_changes,
1029
working_dir=working_dir,
1030
allow_plugins=allow_plugins)
1031
# We distinguish between retcode=None and retcode not passed.
1032
supplied_retcode = kwargs.get('retcode', 0)
1033
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
1034
universal_newlines=kwargs.get('universal_newlines', False),
1037
def start_bzr_subprocess(self, process_args, env_changes=None,
1038
skip_if_plan_to_signal=False,
1040
allow_plugins=False):
1041
"""Start bzr in a subprocess for testing.
1043
This starts a new Python interpreter and runs bzr in there.
1044
This should only be used for tests that have a justifiable need for
1045
this isolation: e.g. they are testing startup time, or signal
1046
handling, or early startup code, etc. Subprocess code can't be
1047
profiled or debugged so easily.
1049
:param process_args: a list of arguments to pass to the bzr executable,
1050
for example `['--version']`.
1051
:param env_changes: A dictionary which lists changes to environment
1052
variables. A value of None will unset the env variable.
1053
The values must be strings. The change will only occur in the
1054
child, so you don't need to fix the environment after running.
1055
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1057
:param allow_plugins: If False (default) pass --no-plugins to bzr.
1059
:returns: Popen object for the started process.
1061
if skip_if_plan_to_signal:
1062
if not getattr(os, 'kill', None):
1063
raise TestSkipped("os.kill not available.")
1065
if env_changes is None:
1069
def cleanup_environment():
1070
for env_var, value in env_changes.iteritems():
1071
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
1073
def restore_environment():
1074
for env_var, value in old_env.iteritems():
1075
osutils.set_or_unset_env(env_var, value)
1077
bzr_path = self.get_bzr_path()
1080
if working_dir is not None:
1081
cwd = osutils.getcwd()
1082
os.chdir(working_dir)
1085
# win32 subprocess doesn't support preexec_fn
1086
# so we will avoid using it on all platforms, just to
1087
# make sure the code path is used, and we don't break on win32
1088
cleanup_environment()
1089
command = [sys.executable, bzr_path]
1090
if not allow_plugins:
1091
command.append('--no-plugins')
1092
command.extend(process_args)
1093
process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1095
restore_environment()
1101
def _popen(self, *args, **kwargs):
1102
"""Place a call to Popen.
1104
Allows tests to override this method to intercept the calls made to
1105
Popen for introspection.
1107
return Popen(*args, **kwargs)
1109
def get_bzr_path(self):
1110
"""Return the path of the 'bzr' executable for this test suite."""
1111
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1112
if not os.path.isfile(bzr_path):
1113
# We are probably installed. Assume sys.argv is the right file
1114
bzr_path = sys.argv[0]
1117
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1118
universal_newlines=False, process_args=None):
1119
"""Finish the execution of process.
1121
:param process: the Popen object returned from start_bzr_subprocess.
1122
:param retcode: The status code that is expected. Defaults to 0. If
1123
None is supplied, the status code is not checked.
1124
:param send_signal: an optional signal to send to the process.
1125
:param universal_newlines: Convert CRLF => LF
1126
:returns: (stdout, stderr)
1128
if send_signal is not None:
1129
os.kill(process.pid, send_signal)
1130
out, err = process.communicate()
1132
if universal_newlines:
1133
out = out.replace('\r\n', '\n')
1134
err = err.replace('\r\n', '\n')
1136
if retcode is not None and retcode != process.returncode:
1137
if process_args is None:
1138
process_args = "(unknown args)"
1139
mutter('Output of bzr %s:\n%s', process_args, out)
1140
mutter('Error for bzr %s:\n%s', process_args, err)
1141
self.fail('Command bzr %s failed with retcode %s != %s'
1142
% (process_args, retcode, process.returncode))
1145
def check_inventory_shape(self, inv, shape):
1146
"""Compare an inventory to a list of expected names.
1148
Fail if they are not precisely equal.
1151
shape = list(shape) # copy
1152
for path, ie in inv.entries():
1153
name = path.replace('\\', '/')
1154
if ie.kind == 'dir':
1161
self.fail("expected paths not found in inventory: %r" % shape)
1163
self.fail("unexpected paths found in inventory: %r" % extras)
1165
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
1166
a_callable=None, *args, **kwargs):
1167
"""Call callable with redirected std io pipes.
1169
Returns the return code."""
1170
if not callable(a_callable):
1171
raise ValueError("a_callable must be callable.")
1173
stdin = StringIO("")
1175
if getattr(self, "_log_file", None) is not None:
1176
stdout = self._log_file
1180
if getattr(self, "_log_file", None is not None):
1181
stderr = self._log_file
1184
real_stdin = sys.stdin
1185
real_stdout = sys.stdout
1186
real_stderr = sys.stderr
1191
return a_callable(*args, **kwargs)
1193
sys.stdout = real_stdout
1194
sys.stderr = real_stderr
1195
sys.stdin = real_stdin
1197
@symbol_versioning.deprecated_method(symbol_versioning.zero_eleven)
1198
def merge(self, branch_from, wt_to):
1199
"""A helper for tests to do a ui-less merge.
1201
This should move to the main library when someone has time to integrate
1204
# minimal ui-less merge.
1205
wt_to.branch.fetch(branch_from)
1206
base_rev = common_ancestor(branch_from.last_revision(),
1207
wt_to.branch.last_revision(),
1208
wt_to.branch.repository)
1209
merge_inner(wt_to.branch, branch_from.basis_tree(),
1210
wt_to.branch.repository.revision_tree(base_rev),
1212
wt_to.add_parent_tree_id(branch_from.last_revision())
1215
BzrTestBase = TestCase
1218
class TestCaseWithMemoryTransport(TestCase):
1219
"""Common test class for tests that do not need disk resources.
1221
Tests that need disk resources should derive from TestCaseWithTransport.
1223
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1225
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1226
a directory which does not exist. This serves to help ensure test isolation
1227
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1228
must exist. However, TestCaseWithMemoryTransport does not offer local
1229
file defaults for the transport in tests, nor does it obey the command line
1230
override, so tests that accidentally write to the common directory should
1238
def __init__(self, methodName='runTest'):
1239
# allow test parameterisation after test construction and before test
1240
# execution. Variables that the parameteriser sets need to be
1241
# ones that are not set by setUp, or setUp will trash them.
1242
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1243
self.vfs_transport_factory = default_transport
1244
self.transport_server = None
1245
self.transport_readonly_server = None
1247
def failUnlessExists(self, path):
1248
"""Fail unless path, which may be abs or relative, exists."""
1249
self.failUnless(osutils.lexists(path))
1251
def failIfExists(self, path):
1252
"""Fail if path, which may be abs or relative, exists."""
1253
self.failIf(osutils.lexists(path))
1255
def get_transport(self):
1256
"""Return a writeable transport for the test scratch space"""
1257
t = get_transport(self.get_url())
1258
self.assertFalse(t.is_readonly())
1261
def get_readonly_transport(self):
1262
"""Return a readonly transport for the test scratch space
1264
This can be used to test that operations which should only need
1265
readonly access in fact do not try to write.
1267
t = get_transport(self.get_readonly_url())
1268
self.assertTrue(t.is_readonly())
1271
def create_transport_readonly_server(self):
1272
"""Create a transport server from class defined at init.
1274
This is mostly a hook for daughter classes.
1276
return self.transport_readonly_server()
1278
def get_readonly_server(self):
1279
"""Get the server instance for the readonly transport
1281
This is useful for some tests with specific servers to do diagnostics.
1283
if self.__readonly_server is None:
1284
if self.transport_readonly_server is None:
1285
# readonly decorator requested
1286
# bring up the server
1287
self.__readonly_server = ReadonlyServer()
1288
self.__readonly_server.setUp(self.get_vfs_only_server())
1290
self.__readonly_server = self.create_transport_readonly_server()
1291
self.__readonly_server.setUp(self.get_vfs_only_server())
1292
self.addCleanup(self.__readonly_server.tearDown)
1293
return self.__readonly_server
1295
def get_readonly_url(self, relpath=None):
1296
"""Get a URL for the readonly transport.
1298
This will either be backed by '.' or a decorator to the transport
1299
used by self.get_url()
1300
relpath provides for clients to get a path relative to the base url.
1301
These should only be downwards relative, not upwards.
1303
base = self.get_readonly_server().get_url()
1304
if relpath is not None:
1305
if not base.endswith('/'):
1307
base = base + relpath
1310
def get_vfs_only_server(self):
1311
"""Get thae vfs only read/write server instance.
1313
This is useful for some tests with specific servers that need
1316
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1317
is no means to override it.
1319
if self.__vfs_server is None:
1320
self.__vfs_server = MemoryServer()
1321
self.__vfs_server.setUp()
1322
self.addCleanup(self.__vfs_server.tearDown)
1323
return self.__vfs_server
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
This is built from the self.transport_server factory. If that is None,
1332
then the self.get_vfs_server is returned.
1334
if self.__server is None:
1335
if self.transport_server is None:
1336
return self.get_vfs_only_server()
1338
# bring up a decorated means of access to the vfs only server.
1339
self.__server = self.transport_server()
1340
self.__server.setUp(self.get_vfs_only_server())
1341
self.addCleanup(self.__server.tearDown)
1342
return self.__server
1344
def _adjust_url(self, base, relpath):
1345
"""Get a URL (or maybe a path) for the readwrite transport.
1347
This will either be backed by '.' or to an equivalent non-file based
1349
relpath provides for clients to get a path relative to the base url.
1350
These should only be downwards relative, not upwards.
1352
if relpath is not None and relpath != '.':
1353
if not base.endswith('/'):
1355
# XXX: Really base should be a url; we did after all call
1356
# get_url()! But sometimes it's just a path (from
1357
# LocalAbspathServer), and it'd be wrong to append urlescaped data
1358
# to a non-escaped local path.
1359
if base.startswith('./') or base.startswith('/'):
1362
base += urlutils.escape(relpath)
1365
def get_url(self, relpath=None):
1366
"""Get a URL (or maybe a path) for the readwrite transport.
1368
This will either be backed by '.' or to an equivalent non-file based
1370
relpath provides for clients to get a path relative to the base url.
1371
These should only be downwards relative, not upwards.
1373
base = self.get_server().get_url()
1374
return self._adjust_url(base, relpath)
1376
def get_vfs_only_url(self, relpath=None):
1377
"""Get a URL (or maybe a path for the plain old vfs transport.
1379
This will never be a smart protocol.
1380
:param relpath: provides for clients to get a path relative to the base
1381
url. These should only be downwards relative, not upwards.
1383
base = self.get_vfs_only_server().get_url()
1384
return self._adjust_url(base, relpath)
1386
def _make_test_root(self):
1387
if TestCaseWithMemoryTransport.TEST_ROOT is not None:
1391
root = u'test%04d.tmp' % i
1395
if e.errno == errno.EEXIST:
1400
# successfully created
1401
TestCaseWithMemoryTransport.TEST_ROOT = osutils.abspath(root)
1403
# make a fake bzr directory there to prevent any tests propagating
1404
# up onto the source directory's real branch
1405
bzrdir.BzrDir.create_standalone_workingtree(
1406
TestCaseWithMemoryTransport.TEST_ROOT)
1408
def makeAndChdirToTestDir(self):
1409
"""Create a temporary directories for this one test.
1411
This must set self.test_home_dir and self.test_dir and chdir to
1414
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1416
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1417
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1418
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1420
def make_branch(self, relpath, format=None):
1421
"""Create a branch on the transport at relpath."""
1422
repo = self.make_repository(relpath, format=format)
1423
return repo.bzrdir.create_branch()
1425
def make_bzrdir(self, relpath, format=None):
1427
# might be a relative or absolute path
1428
maybe_a_url = self.get_url(relpath)
1429
segments = maybe_a_url.rsplit('/', 1)
1430
t = get_transport(maybe_a_url)
1431
if len(segments) > 1 and segments[-1] not in ('', '.'):
1434
except errors.FileExists:
1437
format = bzrlib.bzrdir.BzrDirFormat.get_default_format()
1438
return format.initialize_on_transport(t)
1439
except errors.UninitializableFormat:
1440
raise TestSkipped("Format %s is not initializable." % format)
1442
def make_repository(self, relpath, shared=False, format=None):
1443
"""Create a repository on our default transport at relpath."""
1444
made_control = self.make_bzrdir(relpath, format=format)
1445
return made_control.create_repository(shared=shared)
1447
def make_branch_and_memory_tree(self, relpath, format=None):
1448
"""Create a branch on the default transport and a MemoryTree for it."""
1449
b = self.make_branch(relpath, format=format)
1450
return memorytree.MemoryTree.create_on_branch(b)
1452
def overrideEnvironmentForTesting(self):
1453
os.environ['HOME'] = self.test_home_dir
1454
os.environ['APPDATA'] = self.test_home_dir
1457
super(TestCaseWithMemoryTransport, self).setUp()
1458
self._make_test_root()
1459
_currentdir = os.getcwdu()
1460
def _leaveDirectory():
1461
os.chdir(_currentdir)
1462
self.addCleanup(_leaveDirectory)
1463
self.makeAndChdirToTestDir()
1464
self.overrideEnvironmentForTesting()
1465
self.__readonly_server = None
1466
self.__server = None
1469
class TestCaseInTempDir(TestCaseWithMemoryTransport):
1470
"""Derived class that runs a test within a temporary directory.
1472
This is useful for tests that need to create a branch, etc.
1474
The directory is created in a slightly complex way: for each
1475
Python invocation, a new temporary top-level directory is created.
1476
All test cases create their own directory within that. If the
1477
tests complete successfully, the directory is removed.
1479
InTempDir is an old alias for FunctionalTestCase.
1482
OVERRIDE_PYTHON = 'python'
1484
def check_file_contents(self, filename, expect):
1485
self.log("check contents of file %s" % filename)
1486
contents = file(filename, 'r').read()
1487
if contents != expect:
1488
self.log("expected: %r" % expect)
1489
self.log("actually: %r" % contents)
1490
self.fail("contents of %s not as expected" % filename)
1492
def makeAndChdirToTestDir(self):
1493
"""See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
1495
For TestCaseInTempDir we create a temporary directory based on the test
1496
name and then create two subdirs - test and home under it.
1498
# shorten the name, to avoid test failures due to path length
1499
short_id = self.id().replace('bzrlib.tests.', '') \
1500
.replace('__main__.', '')[-100:]
1501
# it's possible the same test class is run several times for
1502
# parameterized tests, so make sure the names don't collide.
1506
candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
1508
candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
1509
if os.path.exists(candidate_dir):
1513
os.mkdir(candidate_dir)
1514
self.test_home_dir = candidate_dir + '/home'
1515
os.mkdir(self.test_home_dir)
1516
self.test_dir = candidate_dir + '/work'
1517
os.mkdir(self.test_dir)
1518
os.chdir(self.test_dir)
1521
def build_tree(self, shape, line_endings='native', transport=None):
1522
"""Build a test tree according to a pattern.
1524
shape is a sequence of file specifications. If the final
1525
character is '/', a directory is created.
1527
This assumes that all the elements in the tree being built are new.
1529
This doesn't add anything to a branch.
1530
:param line_endings: Either 'binary' or 'native'
1531
in binary mode, exact contents are written
1532
in native mode, the line endings match the
1533
default platform endings.
1535
:param transport: A transport to write to, for building trees on
1536
VFS's. If the transport is readonly or None,
1537
"." is opened automatically.
1539
# It's OK to just create them using forward slashes on windows.
1540
if transport is None or transport.is_readonly():
1541
transport = get_transport(".")
1543
self.assert_(isinstance(name, basestring))
1545
transport.mkdir(urlutils.escape(name[:-1]))
1547
if line_endings == 'binary':
1549
elif line_endings == 'native':
1552
raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
1553
content = "contents of %s%s" % (name.encode('utf-8'), end)
1554
# Technically 'put()' is the right command. However, put
1555
# uses an AtomicFile, which requires an extra rename into place
1556
# As long as the files didn't exist in the past, append() will
1557
# do the same thing as put()
1558
# On jam's machine, make_kernel_like_tree is:
1559
# put: 4.5-7.5s (averaging 6s)
1561
# put_non_atomic: 2.9-4.5s
1562
transport.put_bytes_non_atomic(urlutils.escape(name), content)
1564
def build_tree_contents(self, shape):
1565
build_tree_contents(shape)
1567
def assertFileEqual(self, content, path):
1568
"""Fail if path does not contain 'content'."""
1569
self.failUnless(osutils.lexists(path))
1570
# TODO: jam 20060427 Shouldn't this be 'rb'?
1571
self.assertEqualDiff(content, open(path, 'r').read())
1574
class TestCaseWithTransport(TestCaseInTempDir):
1575
"""A test case that provides get_url and get_readonly_url facilities.
1577
These back onto two transport servers, one for readonly access and one for
1580
If no explicit class is provided for readonly access, a
1581
ReadonlyTransportDecorator is used instead which allows the use of non disk
1582
based read write transports.
1584
If an explicit class is provided for readonly access, that server and the
1585
readwrite one must both define get_url() as resolving to os.getcwd().
1588
def get_vfs_only_server(self):
1589
"""See TestCaseWithMemoryTransport.
1591
This is useful for some tests with specific servers that need
1594
if self.__vfs_server is None:
1595
self.__vfs_server = self.vfs_transport_factory()
1596
self.__vfs_server.setUp()
1597
self.addCleanup(self.__vfs_server.tearDown)
1598
return self.__vfs_server
1600
def make_branch_and_tree(self, relpath, format=None):
1601
"""Create a branch on the transport and a tree locally.
1603
If the transport is not a LocalTransport, the Tree can't be created on
1604
the transport. In that case the working tree is created in the local
1605
directory, and the returned tree's branch and repository will also be
1608
This will fail if the original default transport for this test
1609
case wasn't backed by the working directory, as the branch won't
1610
be on disk for us to open it.
1612
:param format: The BzrDirFormat.
1613
:returns: the WorkingTree.
1615
# TODO: always use the local disk path for the working tree,
1616
# this obviously requires a format that supports branch references
1617
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
1619
b = self.make_branch(relpath, format=format)
1621
return b.bzrdir.create_workingtree()
1622
except errors.NotLocalUrl:
1623
# We can only make working trees locally at the moment. If the
1624
# transport can't support them, then reopen the branch on a local
1625
# transport, and create the working tree there.
1627
# Possibly we should instead keep
1628
# the non-disk-backed branch and create a local checkout?
1629
bd = bzrdir.BzrDir.open(relpath)
1630
return bd.create_workingtree()
1632
def assertIsDirectory(self, relpath, transport):
1633
"""Assert that relpath within transport is a directory.
1635
This may not be possible on all transports; in that case it propagates
1636
a TransportNotPossible.
1639
mode = transport.stat(relpath).st_mode
1640
except errors.NoSuchFile:
1641
self.fail("path %s is not a directory; no such file"
1643
if not stat.S_ISDIR(mode):
1644
self.fail("path %s is not a directory; has mode %#o"
1648
super(TestCaseWithTransport, self).setUp()
1649
self.__vfs_server = None
1652
class ChrootedTestCase(TestCaseWithTransport):
1653
"""A support class that provides readonly urls outside the local namespace.
1655
This is done by checking if self.transport_server is a MemoryServer. if it
1656
is then we are chrooted already, if it is not then an HttpServer is used
1659
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
1660
be used without needed to redo it when a different
1661
subclass is in use ?
1665
super(ChrootedTestCase, self).setUp()
1666
# NB: one HttpServer is taught to decorate properly,
1667
# this hack can be removed.
1668
if not self.vfs_transport_factory == MemoryServer:
1669
self.transport_readonly_server = HttpServer
1672
def filter_suite_by_re(suite, pattern):
1673
result = TestUtil.TestSuite()
1674
filter_re = re.compile(pattern)
1675
for test in iter_suite_tests(suite):
1676
if filter_re.search(test.id()):
1677
result.addTest(test)
1681
def run_suite(suite, name='test', verbose=False, pattern=".*",
1682
stop_on_failure=False, keep_output=False,
1683
transport=None, lsprof_timed=None, bench_history=None):
1684
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
1689
runner = TextTestRunner(stream=sys.stdout,
1691
verbosity=verbosity,
1692
keep_output=keep_output,
1693
bench_history=bench_history)
1694
runner.stop_on_failure=stop_on_failure
1696
suite = filter_suite_by_re(suite, pattern)
1697
result = runner.run(suite)
1698
return result.wasSuccessful()
1701
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
1704
test_suite_factory=None,
1706
bench_history=None):
1707
"""Run the whole test suite under the enhanced runner"""
1708
# XXX: Very ugly way to do this...
1709
# Disable warning about old formats because we don't want it to disturb
1710
# any blackbox tests.
1711
from bzrlib import repository
1712
repository._deprecation_warning_done = True
1714
global default_transport
1715
if transport is None:
1716
transport = default_transport
1717
old_transport = default_transport
1718
default_transport = transport
1720
if test_suite_factory is None:
1721
suite = test_suite()
1723
suite = test_suite_factory()
1724
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
1725
stop_on_failure=stop_on_failure, keep_output=keep_output,
1726
transport=transport,
1727
lsprof_timed=lsprof_timed,
1728
bench_history=bench_history)
1730
default_transport = old_transport
1734
"""Build and return TestSuite for the whole of bzrlib.
1736
This function can be replaced if you need to change the default test
1737
suite on a global basis, but it is not encouraged.
1740
'bzrlib.tests.test_ancestry',
1741
'bzrlib.tests.test_api',
1742
'bzrlib.tests.test_atomicfile',
1743
'bzrlib.tests.test_bad_files',
1744
'bzrlib.tests.test_branch',
1745
'bzrlib.tests.test_bundle',
1746
'bzrlib.tests.test_bzrdir',
1747
'bzrlib.tests.test_cache_utf8',
1748
'bzrlib.tests.test_command',
1749
'bzrlib.tests.test_commit',
1750
'bzrlib.tests.test_commit_merge',
1751
'bzrlib.tests.test_config',
1752
'bzrlib.tests.test_conflicts',
1753
'bzrlib.tests.test_decorators',
1754
'bzrlib.tests.test_diff',
1755
'bzrlib.tests.test_doc_generate',
1756
'bzrlib.tests.test_errors',
1757
'bzrlib.tests.test_escaped_store',
1758
'bzrlib.tests.test_fetch',
1759
'bzrlib.tests.test_ftp_transport',
1760
'bzrlib.tests.test_generate_ids',
1761
'bzrlib.tests.test_gpg',
1762
'bzrlib.tests.test_graph',
1763
'bzrlib.tests.test_hashcache',
1764
'bzrlib.tests.test_http',
1765
'bzrlib.tests.test_http_response',
1766
'bzrlib.tests.test_identitymap',
1767
'bzrlib.tests.test_ignores',
1768
'bzrlib.tests.test_inv',
1769
'bzrlib.tests.test_knit',
1770
'bzrlib.tests.test_lazy_import',
1771
'bzrlib.tests.test_lazy_regex',
1772
'bzrlib.tests.test_lockdir',
1773
'bzrlib.tests.test_lockable_files',
1774
'bzrlib.tests.test_log',
1775
'bzrlib.tests.test_memorytree',
1776
'bzrlib.tests.test_merge',
1777
'bzrlib.tests.test_merge3',
1778
'bzrlib.tests.test_merge_core',
1779
'bzrlib.tests.test_missing',
1780
'bzrlib.tests.test_msgeditor',
1781
'bzrlib.tests.test_nonascii',
1782
'bzrlib.tests.test_options',
1783
'bzrlib.tests.test_osutils',
1784
'bzrlib.tests.test_patch',
1785
'bzrlib.tests.test_patches',
1786
'bzrlib.tests.test_permissions',
1787
'bzrlib.tests.test_plugins',
1788
'bzrlib.tests.test_progress',
1789
'bzrlib.tests.test_reconcile',
1790
'bzrlib.tests.test_registry',
1791
'bzrlib.tests.test_remote',
1792
'bzrlib.tests.test_repository',
1793
'bzrlib.tests.test_revert',
1794
'bzrlib.tests.test_revision',
1795
'bzrlib.tests.test_revisionnamespaces',
1796
'bzrlib.tests.test_revisiontree',
1797
'bzrlib.tests.test_rio',
1798
'bzrlib.tests.test_sampler',
1799
'bzrlib.tests.test_selftest',
1800
'bzrlib.tests.test_setup',
1801
'bzrlib.tests.test_sftp_transport',
1802
'bzrlib.tests.test_smart',
1803
'bzrlib.tests.test_smart_add',
1804
'bzrlib.tests.test_smart_transport',
1805
'bzrlib.tests.test_source',
1806
'bzrlib.tests.test_status',
1807
'bzrlib.tests.test_store',
1808
'bzrlib.tests.test_symbol_versioning',
1809
'bzrlib.tests.test_testament',
1810
'bzrlib.tests.test_textfile',
1811
'bzrlib.tests.test_textmerge',
1812
'bzrlib.tests.test_trace',
1813
'bzrlib.tests.test_transactions',
1814
'bzrlib.tests.test_transform',
1815
'bzrlib.tests.test_transport',
1816
'bzrlib.tests.test_tree',
1817
'bzrlib.tests.test_treebuilder',
1818
'bzrlib.tests.test_tsort',
1819
'bzrlib.tests.test_tuned_gzip',
1820
'bzrlib.tests.test_ui',
1821
'bzrlib.tests.test_upgrade',
1822
'bzrlib.tests.test_urlutils',
1823
'bzrlib.tests.test_versionedfile',
1824
'bzrlib.tests.test_version',
1825
'bzrlib.tests.test_version_info',
1826
'bzrlib.tests.test_weave',
1827
'bzrlib.tests.test_whitebox',
1828
'bzrlib.tests.test_workingtree',
1829
'bzrlib.tests.test_wsgi',
1830
'bzrlib.tests.test_xml',
1832
test_transport_implementations = [
1833
'bzrlib.tests.test_transport_implementations',
1834
'bzrlib.tests.test_read_bundle',
1836
suite = TestUtil.TestSuite()
1837
loader = TestUtil.TestLoader()
1838
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1839
from bzrlib.transport import TransportTestProviderAdapter
1840
adapter = TransportTestProviderAdapter()
1841
adapt_modules(test_transport_implementations, adapter, loader, suite)
1842
for package in packages_to_test():
1843
suite.addTest(package.test_suite())
1844
for m in MODULES_TO_TEST:
1845
suite.addTest(loader.loadTestsFromModule(m))
1846
for m in MODULES_TO_DOCTEST:
1848
suite.addTest(doctest.DocTestSuite(m))
1849
except ValueError, e:
1850
print '**failed to get doctest for: %s\n%s' %(m,e)
1852
for name, plugin in bzrlib.plugin.all_plugins().items():
1853
if getattr(plugin, 'test_suite', None) is not None:
1854
suite.addTest(plugin.test_suite())
1858
def adapt_modules(mods_list, adapter, loader, suite):
1859
"""Adapt the modules in mods_list using adapter and add to suite."""
1860
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1861
suite.addTests(adapter.adapt(test))
added
removed
bzrlib/tests/__init__.py
