1
# Copyright (C) 2005, 2006 by 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
38
from subprocess import Popen, PIPE
46
import bzrlib.bzrdir as bzrdir
47
import bzrlib.commands
48
import bzrlib.bundle.serializer
49
import bzrlib.errors as errors
50
import bzrlib.inventory
51
import bzrlib.iterablefile
56
# lsprof not available
58
from bzrlib.merge import merge_inner
61
import bzrlib.osutils as osutils
63
import bzrlib.progress as progress
64
from bzrlib.revision import common_ancestor
67
from bzrlib.transport import get_transport
68
import bzrlib.transport
69
from bzrlib.transport.local import LocalRelpathServer
70
from bzrlib.transport.readonly import ReadonlyServer
71
from bzrlib.trace import mutter
72
from bzrlib.tests.TestUtil import TestLoader, TestSuite
73
from bzrlib.tests.treeshape import build_tree_contents
74
import bzrlib.urlutils as urlutils
75
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
77
default_transport = LocalRelpathServer
80
MODULES_TO_DOCTEST = [
82
bzrlib.bundle.serializer,
95
def packages_to_test():
96
"""Return a list of packages to test.
98
The packages are not globally imported so that import failures are
99
triggered when running selftest, not when importing the command.
102
import bzrlib.tests.blackbox
103
import bzrlib.tests.branch_implementations
104
import bzrlib.tests.bzrdir_implementations
105
import bzrlib.tests.interrepository_implementations
106
import bzrlib.tests.interversionedfile_implementations
107
import bzrlib.tests.repository_implementations
108
import bzrlib.tests.revisionstore_implementations
109
import bzrlib.tests.workingtree_implementations
112
bzrlib.tests.blackbox,
113
bzrlib.tests.branch_implementations,
114
bzrlib.tests.bzrdir_implementations,
115
bzrlib.tests.interrepository_implementations,
116
bzrlib.tests.interversionedfile_implementations,
117
bzrlib.tests.repository_implementations,
118
bzrlib.tests.revisionstore_implementations,
119
bzrlib.tests.workingtree_implementations,
123
class _MyResult(unittest._TextTestResult):
124
"""Custom TestResult.
126
Shows output in a different format, including displaying runtime for tests.
130
def __init__(self, stream, descriptions, verbosity, pb=None):
131
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
134
def extractBenchmarkTime(self, testCase):
135
"""Add a benchmark time for the current test case."""
136
self._benchmarkTime = getattr(testCase, "_benchtime", None)
138
def _elapsedTestTimeString(self):
139
"""Return a time string for the overall time the current test has taken."""
140
return self._formatTime(time.time() - self._start_time)
142
def _testTimeString(self):
143
if self._benchmarkTime is not None:
145
self._formatTime(self._benchmarkTime),
146
self._elapsedTestTimeString())
148
return " %s" % self._elapsedTestTimeString()
150
def _formatTime(self, seconds):
151
"""Format seconds as milliseconds with leading spaces."""
152
return "%5dms" % (1000 * seconds)
154
def _ellipsise_unimportant_words(self, a_string, final_width,
156
"""Add ellipses (sp?) for overly long strings.
158
:param keep_start: If true preserve the start of a_string rather
162
if len(a_string) > final_width:
163
result = a_string[:final_width-3] + '...'
167
if len(a_string) > final_width:
168
result = '...' + a_string[3-final_width:]
171
return result.ljust(final_width)
173
def startTest(self, test):
174
unittest.TestResult.startTest(self, test)
175
# In a short description, the important words are in
176
# the beginning, but in an id, the important words are
178
SHOW_DESCRIPTIONS = False
180
if not self.showAll and self.dots and self.pb is not None:
183
final_width = osutils.terminal_width()
184
final_width = final_width - 15 - 8
186
if SHOW_DESCRIPTIONS:
187
what = test.shortDescription()
189
what = self._ellipsise_unimportant_words(what, final_width, keep_start=True)
192
if what.startswith('bzrlib.tests.'):
194
what = self._ellipsise_unimportant_words(what, final_width)
196
self.stream.write(what)
197
elif self.dots and self.pb is not None:
198
self.pb.update(what, self.testsRun - 1, None)
200
self._recordTestStartTime()
202
def _recordTestStartTime(self):
203
"""Record that a test has started."""
204
self._start_time = time.time()
206
def addError(self, test, err):
207
if isinstance(err[1], TestSkipped):
208
return self.addSkipped(test, err)
209
unittest.TestResult.addError(self, test, err)
210
self.extractBenchmarkTime(test)
212
self.stream.writeln("ERROR %s" % self._testTimeString())
213
elif self.dots and self.pb is None:
214
self.stream.write('E')
216
self.pb.update(self._ellipsise_unimportant_words('ERROR', 13), self.testsRun, None)
221
def addFailure(self, test, err):
222
unittest.TestResult.addFailure(self, test, err)
223
self.extractBenchmarkTime(test)
225
self.stream.writeln(" FAIL %s" % self._testTimeString())
226
elif self.dots and self.pb is None:
227
self.stream.write('F')
229
self.pb.update(self._ellipsise_unimportant_words('FAIL', 13), self.testsRun, None)
234
def addSuccess(self, test):
235
self.extractBenchmarkTime(test)
237
self.stream.writeln(' OK %s' % self._testTimeString())
238
for bench_called, stats in getattr(test, '_benchcalls', []):
239
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
240
stats.pprint(file=self.stream)
241
elif self.dots and self.pb is None:
242
self.stream.write('~')
244
self.pb.update(self._ellipsise_unimportant_words('OK', 13), self.testsRun, None)
246
unittest.TestResult.addSuccess(self, test)
248
def addSkipped(self, test, skip_excinfo):
249
self.extractBenchmarkTime(test)
251
print >>self.stream, ' SKIP %s' % self._testTimeString()
252
print >>self.stream, ' %s' % skip_excinfo[1]
253
elif self.dots and self.pb is None:
254
self.stream.write('S')
256
self.pb.update(self._ellipsise_unimportant_words('SKIP', 13), self.testsRun, None)
258
# seems best to treat this as success from point-of-view of unittest
259
# -- it actually does nothing so it barely matters :)
260
unittest.TestResult.addSuccess(self, test)
262
def printErrorList(self, flavour, errors):
263
for test, err in errors:
264
self.stream.writeln(self.separator1)
265
self.stream.writeln("%s: %s" % (flavour, self.getDescription(test)))
266
if getattr(test, '_get_log', None) is not None:
268
print >>self.stream, \
269
('vvvv[log from %s]' % test.id()).ljust(78,'-')
270
print >>self.stream, test._get_log()
271
print >>self.stream, \
272
('^^^^[log from %s]' % test.id()).ljust(78,'-')
273
self.stream.writeln(self.separator2)
274
self.stream.writeln("%s" % err)
277
class TextTestRunner(object):
278
stop_on_failure = False
286
self.stream = unittest._WritelnDecorator(stream)
287
self.descriptions = descriptions
288
self.verbosity = verbosity
289
self.keep_output = keep_output
292
def _makeResult(self):
293
result = _MyResult(self.stream,
297
result.stop_early = self.stop_on_failure
301
"Run the given test case or test suite."
302
result = self._makeResult()
303
startTime = time.time()
304
if self.pb is not None:
305
self.pb.update('Running tests', 0, test.countTestCases())
307
stopTime = time.time()
308
timeTaken = stopTime - startTime
310
self.stream.writeln(result.separator2)
311
run = result.testsRun
312
self.stream.writeln("Ran %d test%s in %.3fs" %
313
(run, run != 1 and "s" or "", timeTaken))
314
self.stream.writeln()
315
if not result.wasSuccessful():
316
self.stream.write("FAILED (")
317
failed, errored = map(len, (result.failures, result.errors))
319
self.stream.write("failures=%d" % failed)
321
if failed: self.stream.write(", ")
322
self.stream.write("errors=%d" % errored)
323
self.stream.writeln(")")
325
self.stream.writeln("OK")
326
if self.pb is not None:
327
self.pb.update('Cleaning up', 0, 1)
328
# This is still a little bogus,
329
# but only a little. Folk not using our testrunner will
330
# have to delete their temp directories themselves.
331
test_root = TestCaseInTempDir.TEST_ROOT
332
if result.wasSuccessful() or not self.keep_output:
333
if test_root is not None:
334
# If LANG=C we probably have created some bogus paths
335
# which rmtree(unicode) will fail to delete
336
# so make sure we are using rmtree(str) to delete everything
337
osutils.rmtree(test_root.encode(
338
sys.getfilesystemencoding()))
340
if self.pb is not None:
341
self.pb.note("Failed tests working directories are in '%s'\n",
345
"Failed tests working directories are in '%s'\n" %
347
TestCaseInTempDir.TEST_ROOT = None
348
if self.pb is not None:
353
def iter_suite_tests(suite):
354
"""Return all tests in a suite, recursing through nested suites"""
355
for item in suite._tests:
356
if isinstance(item, unittest.TestCase):
358
elif isinstance(item, unittest.TestSuite):
359
for r in iter_suite_tests(item):
362
raise Exception('unknown object %r inside test suite %r'
366
class TestSkipped(Exception):
367
"""Indicates that a test was intentionally skipped, rather than failing."""
371
class CommandFailed(Exception):
375
class StringIOWrapper(object):
376
"""A wrapper around cStringIO which just adds an encoding attribute.
378
Internally we can check sys.stdout to see what the output encoding
379
should be. However, cStringIO has no encoding attribute that we can
380
set. So we wrap it instead.
385
def __init__(self, s=None):
387
self.__dict__['_cstring'] = StringIO(s)
389
self.__dict__['_cstring'] = StringIO()
391
def __getattr__(self, name, getattr=getattr):
392
return getattr(self.__dict__['_cstring'], name)
394
def __setattr__(self, name, val):
395
if name == 'encoding':
396
self.__dict__['encoding'] = val
398
return setattr(self._cstring, name, val)
401
class TestCase(unittest.TestCase):
402
"""Base class for bzr unit tests.
404
Tests that need access to disk resources should subclass
405
TestCaseInTempDir not TestCase.
407
Error and debug log messages are redirected from their usual
408
location into a temporary file, the contents of which can be
409
retrieved by _get_log(). We use a real OS file, not an in-memory object,
410
so that it can also capture file IO. When the test completes this file
411
is read into memory and removed from disk.
413
There are also convenience functions to invoke bzr's command-line
414
routine, and to build and check bzr trees.
416
In addition to the usual method of overriding tearDown(), this class also
417
allows subclasses to register functions into the _cleanups list, which is
418
run in order as the object is torn down. It's less likely this will be
419
accidentally overlooked.
422
_log_file_name = None
424
# record lsprof data when performing benchmark calls.
425
_gather_lsprof_in_benchmarks = False
427
def __init__(self, methodName='testMethod'):
428
super(TestCase, self).__init__(methodName)
432
unittest.TestCase.setUp(self)
433
self._cleanEnvironment()
434
bzrlib.trace.disable_default_logging()
436
self._benchcalls = []
437
self._benchtime = None
439
def _ndiff_strings(self, a, b):
440
"""Return ndiff between two strings containing lines.
442
A trailing newline is added if missing to make the strings
444
if b and b[-1] != '\n':
446
if a and a[-1] != '\n':
448
difflines = difflib.ndiff(a.splitlines(True),
450
linejunk=lambda x: False,
451
charjunk=lambda x: False)
452
return ''.join(difflines)
454
def assertEqualDiff(self, a, b, message=None):
455
"""Assert two texts are equal, if not raise an exception.
457
This is intended for use with multi-line strings where it can
458
be hard to find the differences by eye.
460
# TODO: perhaps override assertEquals to call this for strings?
464
message = "texts not equal:\n"
465
raise AssertionError(message +
466
self._ndiff_strings(a, b))
468
def assertEqualMode(self, mode, mode_test):
469
self.assertEqual(mode, mode_test,
470
'mode mismatch %o != %o' % (mode, mode_test))
472
def assertStartsWith(self, s, prefix):
473
if not s.startswith(prefix):
474
raise AssertionError('string %r does not start with %r' % (s, prefix))
476
def assertEndsWith(self, s, suffix):
477
"""Asserts that s ends with suffix."""
478
if not s.endswith(suffix):
479
raise AssertionError('string %r does not end with %r' % (s, suffix))
481
def assertContainsRe(self, haystack, needle_re):
482
"""Assert that a contains something matching a regular expression."""
483
if not re.search(needle_re, haystack):
484
raise AssertionError('pattern "%s" not found in "%s"'
485
% (needle_re, haystack))
487
def assertNotContainsRe(self, haystack, needle_re):
488
"""Assert that a does not match a regular expression"""
489
if re.search(needle_re, haystack):
490
raise AssertionError('pattern "%s" found in "%s"'
491
% (needle_re, haystack))
493
def assertSubset(self, sublist, superlist):
494
"""Assert that every entry in sublist is present in superlist."""
496
for entry in sublist:
497
if entry not in superlist:
498
missing.append(entry)
500
raise AssertionError("value(s) %r not present in container %r" %
501
(missing, superlist))
503
def assertIs(self, left, right):
504
if not (left is right):
505
raise AssertionError("%r is not %r." % (left, right))
507
def assertTransportMode(self, transport, path, mode):
508
"""Fail if a path does not have mode mode.
510
If modes are not supported on this transport, the assertion is ignored.
512
if not transport._can_roundtrip_unix_modebits():
514
path_stat = transport.stat(path)
515
actual_mode = stat.S_IMODE(path_stat.st_mode)
516
self.assertEqual(mode, actual_mode,
517
'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
519
def assertIsInstance(self, obj, kls):
520
"""Fail if obj is not an instance of kls"""
521
if not isinstance(obj, kls):
522
self.fail("%r is an instance of %s rather than %s" % (
523
obj, obj.__class__, kls))
525
def _startLogFile(self):
526
"""Send bzr and test log messages to a temporary file.
528
The file is removed as the test is torn down.
530
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
531
encoder, decoder, stream_reader, stream_writer = codecs.lookup('UTF-8')
532
self._log_file = stream_writer(os.fdopen(fileno, 'w+'))
533
self._log_nonce = bzrlib.trace.enable_test_log(self._log_file)
534
self._log_file_name = name
535
self.addCleanup(self._finishLogFile)
537
def _finishLogFile(self):
538
"""Finished with the log file.
540
Read contents into memory, close, and delete.
542
bzrlib.trace.disable_test_log(self._log_nonce)
543
self._log_file.seek(0)
544
self._log_contents = self._log_file.read()
545
self._log_file.close()
546
os.remove(self._log_file_name)
547
self._log_file = self._log_file_name = None
549
def addCleanup(self, callable):
550
"""Arrange to run a callable when this case is torn down.
552
Callables are run in the reverse of the order they are registered,
553
ie last-in first-out.
555
if callable in self._cleanups:
556
raise ValueError("cleanup function %r already registered on %s"
558
self._cleanups.append(callable)
560
def _cleanEnvironment(self):
563
'APPDATA': os.getcwd(),
568
self.addCleanup(self._restoreEnvironment)
569
for name, value in new_env.iteritems():
570
self._captureVar(name, value)
573
def _captureVar(self, name, newvalue):
574
"""Set an environment variable, preparing it to be reset when finished."""
575
self.__old_env[name] = os.environ.get(name, None)
577
if name in os.environ:
580
os.environ[name] = newvalue
583
def _restoreVar(name, value):
585
if name in os.environ:
588
os.environ[name] = value
590
def _restoreEnvironment(self):
591
for name, value in self.__old_env.iteritems():
592
self._restoreVar(name, value)
596
unittest.TestCase.tearDown(self)
598
def time(self, callable, *args, **kwargs):
599
"""Run callable and accrue the time it takes to the benchmark time.
601
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
602
this will cause lsprofile statistics to be gathered and stored in
605
if self._benchtime is None:
609
if not self._gather_lsprof_in_benchmarks:
610
return callable(*args, **kwargs)
612
# record this benchmark
613
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
615
self._benchcalls.append(((callable, args, kwargs), stats))
618
self._benchtime += time.time() - start
620
def _runCleanups(self):
621
"""Run registered cleanup functions.
623
This should only be called from TestCase.tearDown.
625
# TODO: Perhaps this should keep running cleanups even if
627
for cleanup_fn in reversed(self._cleanups):
630
def log(self, *args):
634
"""Return as a string the log for this test"""
635
if self._log_file_name:
636
return open(self._log_file_name).read()
638
return self._log_contents
639
# TODO: Delete the log after it's been read in
641
def capture(self, cmd, retcode=0):
642
"""Shortcut that splits cmd into words, runs, and returns stdout"""
643
return self.run_bzr_captured(cmd.split(), retcode=retcode)[0]
645
def run_bzr_captured(self, argv, retcode=0, encoding=None, stdin=None):
646
"""Invoke bzr and return (stdout, stderr).
648
Useful for code that wants to check the contents of the
649
output, the way error messages are presented, etc.
651
This should be the main method for tests that want to exercise the
652
overall behavior of the bzr application (rather than a unit test
653
or a functional test of the library.)
655
Much of the old code runs bzr by forking a new copy of Python, but
656
that is slower, harder to debug, and generally not necessary.
658
This runs bzr through the interface that catches and reports
659
errors, and with logging set to something approximating the
660
default, so that error reporting can be checked.
662
:param argv: arguments to invoke bzr
663
:param retcode: expected return code, or None for don't-care.
664
:param encoding: encoding for sys.stdout and sys.stderr
665
:param stdin: A string to be used as stdin for the command.
668
encoding = bzrlib.user_encoding
669
if stdin is not None:
670
stdin = StringIO(stdin)
671
stdout = StringIOWrapper()
672
stderr = StringIOWrapper()
673
stdout.encoding = encoding
674
stderr.encoding = encoding
676
self.log('run bzr: %r', argv)
677
# FIXME: don't call into logging here
678
handler = logging.StreamHandler(stderr)
679
handler.setLevel(logging.INFO)
680
logger = logging.getLogger('')
681
logger.addHandler(handler)
682
old_ui_factory = bzrlib.ui.ui_factory
683
bzrlib.ui.ui_factory = bzrlib.tests.blackbox.TestUIFactory(
686
bzrlib.ui.ui_factory.stdin = stdin
688
result = self.apply_redirected(stdin, stdout, stderr,
689
bzrlib.commands.run_bzr_catch_errors,
692
logger.removeHandler(handler)
693
bzrlib.ui.ui_factory = old_ui_factory
695
out = stdout.getvalue()
696
err = stderr.getvalue()
698
self.log('output:\n%r', out)
700
self.log('errors:\n%r', err)
701
if retcode is not None:
702
self.assertEquals(retcode, result)
705
def run_bzr(self, *args, **kwargs):
706
"""Invoke bzr, as if it were run from the command line.
708
This should be the main method for tests that want to exercise the
709
overall behavior of the bzr application (rather than a unit test
710
or a functional test of the library.)
712
This sends the stdout/stderr results into the test's log,
713
where it may be useful for debugging. See also run_captured.
715
:param stdin: A string to be used as stdin for the command.
717
retcode = kwargs.pop('retcode', 0)
718
encoding = kwargs.pop('encoding', None)
719
stdin = kwargs.pop('stdin', None)
720
return self.run_bzr_captured(args, retcode=retcode, encoding=encoding, stdin=stdin)
722
def run_bzr_decode(self, *args, **kwargs):
723
if kwargs.has_key('encoding'):
724
encoding = kwargs['encoding']
726
encoding = bzrlib.user_encoding
727
return self.run_bzr(*args, **kwargs)[0].decode(encoding)
729
def run_bzr_subprocess(self, *args, **kwargs):
730
"""Run bzr in a subprocess for testing.
732
This starts a new Python interpreter and runs bzr in there.
733
This should only be used for tests that have a justifiable need for
734
this isolation: e.g. they are testing startup time, or signal
735
handling, or early startup code, etc. Subprocess code can't be
736
profiled or debugged so easily.
738
:param retcode: The status code that is expected. Defaults to 0. If
739
None is supplied, the status code is not checked.
741
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
743
process = Popen([sys.executable, bzr_path]+args, stdout=PIPE,
745
out = process.stdout.read()
746
err = process.stderr.read()
747
retcode = process.wait()
748
supplied_retcode = kwargs.get('retcode', 0)
749
if supplied_retcode is not None:
750
assert supplied_retcode == retcode
753
def check_inventory_shape(self, inv, shape):
754
"""Compare an inventory to a list of expected names.
756
Fail if they are not precisely equal.
759
shape = list(shape) # copy
760
for path, ie in inv.entries():
761
name = path.replace('\\', '/')
769
self.fail("expected paths not found in inventory: %r" % shape)
771
self.fail("unexpected paths found in inventory: %r" % extras)
773
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
774
a_callable=None, *args, **kwargs):
775
"""Call callable with redirected std io pipes.
777
Returns the return code."""
778
if not callable(a_callable):
779
raise ValueError("a_callable must be callable.")
783
if getattr(self, "_log_file", None) is not None:
784
stdout = self._log_file
788
if getattr(self, "_log_file", None is not None):
789
stderr = self._log_file
792
real_stdin = sys.stdin
793
real_stdout = sys.stdout
794
real_stderr = sys.stderr
799
return a_callable(*args, **kwargs)
801
sys.stdout = real_stdout
802
sys.stderr = real_stderr
803
sys.stdin = real_stdin
805
def merge(self, branch_from, wt_to):
806
"""A helper for tests to do a ui-less merge.
808
This should move to the main library when someone has time to integrate
811
# minimal ui-less merge.
812
wt_to.branch.fetch(branch_from)
813
base_rev = common_ancestor(branch_from.last_revision(),
814
wt_to.branch.last_revision(),
815
wt_to.branch.repository)
816
merge_inner(wt_to.branch, branch_from.basis_tree(),
817
wt_to.branch.repository.revision_tree(base_rev),
819
wt_to.add_pending_merge(branch_from.last_revision())
822
BzrTestBase = TestCase
825
class TestCaseInTempDir(TestCase):
826
"""Derived class that runs a test within a temporary directory.
828
This is useful for tests that need to create a branch, etc.
830
The directory is created in a slightly complex way: for each
831
Python invocation, a new temporary top-level directory is created.
832
All test cases create their own directory within that. If the
833
tests complete successfully, the directory is removed.
835
InTempDir is an old alias for FunctionalTestCase.
840
OVERRIDE_PYTHON = 'python'
842
def check_file_contents(self, filename, expect):
843
self.log("check contents of file %s" % filename)
844
contents = file(filename, 'r').read()
845
if contents != expect:
846
self.log("expected: %r" % expect)
847
self.log("actually: %r" % contents)
848
self.fail("contents of %s not as expected" % filename)
850
def _make_test_root(self):
851
if TestCaseInTempDir.TEST_ROOT is not None:
855
root = u'test%04d.tmp' % i
859
if e.errno == errno.EEXIST:
864
# successfully created
865
TestCaseInTempDir.TEST_ROOT = osutils.abspath(root)
867
# make a fake bzr directory there to prevent any tests propagating
868
# up onto the source directory's real branch
869
bzrdir.BzrDir.create_standalone_workingtree(TestCaseInTempDir.TEST_ROOT)
872
super(TestCaseInTempDir, self).setUp()
873
self._make_test_root()
874
_currentdir = os.getcwdu()
875
# shorten the name, to avoid test failures due to path length
876
short_id = self.id().replace('bzrlib.tests.', '') \
877
.replace('__main__.', '')[-100:]
878
# it's possible the same test class is run several times for
879
# parameterized tests, so make sure the names don't collide.
883
candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
885
candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
886
if os.path.exists(candidate_dir):
890
self.test_dir = candidate_dir
891
os.mkdir(self.test_dir)
892
os.chdir(self.test_dir)
894
os.environ['HOME'] = self.test_dir
895
os.environ['APPDATA'] = self.test_dir
896
def _leaveDirectory():
897
os.chdir(_currentdir)
898
self.addCleanup(_leaveDirectory)
900
def build_tree(self, shape, line_endings='native', transport=None):
901
"""Build a test tree according to a pattern.
903
shape is a sequence of file specifications. If the final
904
character is '/', a directory is created.
906
This doesn't add anything to a branch.
907
:param line_endings: Either 'binary' or 'native'
908
in binary mode, exact contents are written
909
in native mode, the line endings match the
910
default platform endings.
912
:param transport: A transport to write to, for building trees on
913
VFS's. If the transport is readonly or None,
914
"." is opened automatically.
916
# XXX: It's OK to just create them using forward slashes on windows?
917
if transport is None or transport.is_readonly():
918
transport = get_transport(".")
920
self.assert_(isinstance(name, basestring))
922
transport.mkdir(urlutils.escape(name[:-1]))
924
if line_endings == 'binary':
926
elif line_endings == 'native':
929
raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
930
content = "contents of %s%s" % (name.encode('utf-8'), end)
931
transport.put(urlutils.escape(name), StringIO(content))
933
def build_tree_contents(self, shape):
934
build_tree_contents(shape)
936
def failUnlessExists(self, path):
937
"""Fail unless path, which may be abs or relative, exists."""
938
self.failUnless(osutils.lexists(path))
940
def failIfExists(self, path):
941
"""Fail if path, which may be abs or relative, exists."""
942
self.failIf(osutils.lexists(path))
944
def assertFileEqual(self, content, path):
945
"""Fail if path does not contain 'content'."""
946
self.failUnless(osutils.lexists(path))
947
# TODO: jam 20060427 Shouldn't this be 'rb'?
948
self.assertEqualDiff(content, open(path, 'r').read())
951
class TestCaseWithTransport(TestCaseInTempDir):
952
"""A test case that provides get_url and get_readonly_url facilities.
954
These back onto two transport servers, one for readonly access and one for
957
If no explicit class is provided for readonly access, a
958
ReadonlyTransportDecorator is used instead which allows the use of non disk
959
based read write transports.
961
If an explicit class is provided for readonly access, that server and the
962
readwrite one must both define get_url() as resolving to os.getcwd().
965
def __init__(self, methodName='testMethod'):
966
super(TestCaseWithTransport, self).__init__(methodName)
967
self.__readonly_server = None
969
self.transport_server = default_transport
970
self.transport_readonly_server = None
972
def get_readonly_url(self, relpath=None):
973
"""Get a URL for the readonly transport.
975
This will either be backed by '.' or a decorator to the transport
976
used by self.get_url()
977
relpath provides for clients to get a path relative to the base url.
978
These should only be downwards relative, not upwards.
980
base = self.get_readonly_server().get_url()
981
if relpath is not None:
982
if not base.endswith('/'):
984
base = base + relpath
987
def get_readonly_server(self):
988
"""Get the server instance for the readonly transport
990
This is useful for some tests with specific servers to do diagnostics.
992
if self.__readonly_server is None:
993
if self.transport_readonly_server is None:
994
# readonly decorator requested
995
# bring up the server
997
self.__readonly_server = ReadonlyServer()
998
self.__readonly_server.setUp(self.__server)
1000
self.__readonly_server = self.transport_readonly_server()
1001
self.__readonly_server.setUp()
1002
self.addCleanup(self.__readonly_server.tearDown)
1003
return self.__readonly_server
1005
def get_server(self):
1006
"""Get the read/write server instance.
1008
This is useful for some tests with specific servers that need
1011
if self.__server is None:
1012
self.__server = self.transport_server()
1013
self.__server.setUp()
1014
self.addCleanup(self.__server.tearDown)
1015
return self.__server
1017
def get_url(self, relpath=None):
1018
"""Get a URL for the readwrite transport.
1020
This will either be backed by '.' or to an equivalent non-file based
1022
relpath provides for clients to get a path relative to the base url.
1023
These should only be downwards relative, not upwards.
1025
base = self.get_server().get_url()
1026
if relpath is not None and relpath != '.':
1027
if not base.endswith('/'):
1029
base = base + urlutils.escape(relpath)
1032
def get_transport(self):
1033
"""Return a writeable transport for the test scratch space"""
1034
t = get_transport(self.get_url())
1035
self.assertFalse(t.is_readonly())
1038
def get_readonly_transport(self):
1039
"""Return a readonly transport for the test scratch space
1041
This can be used to test that operations which should only need
1042
readonly access in fact do not try to write.
1044
t = get_transport(self.get_readonly_url())
1045
self.assertTrue(t.is_readonly())
1048
def make_branch(self, relpath, format=None):
1049
"""Create a branch on the transport at relpath."""
1050
repo = self.make_repository(relpath, format=format)
1051
return repo.bzrdir.create_branch()
1053
def make_bzrdir(self, relpath, format=None):
1055
url = self.get_url(relpath)
1056
mutter('relpath %r => url %r', relpath, url)
1057
segments = url.split('/')
1058
if segments and segments[-1] not in ('', '.'):
1059
parent = '/'.join(segments[:-1])
1060
t = get_transport(parent)
1062
t.mkdir(segments[-1])
1063
except errors.FileExists:
1066
format=bzrlib.bzrdir.BzrDirFormat.get_default_format()
1067
# FIXME: make this use a single transport someday. RBC 20060418
1068
return format.initialize_on_transport(get_transport(relpath))
1069
except errors.UninitializableFormat:
1070
raise TestSkipped("Format %s is not initializable." % format)
1072
def make_repository(self, relpath, shared=False, format=None):
1073
"""Create a repository on our default transport at relpath."""
1074
made_control = self.make_bzrdir(relpath, format=format)
1075
return made_control.create_repository(shared=shared)
1077
def make_branch_and_tree(self, relpath, format=None):
1078
"""Create a branch on the transport and a tree locally.
1082
# TODO: always use the local disk path for the working tree,
1083
# this obviously requires a format that supports branch references
1084
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
1086
b = self.make_branch(relpath, format=format)
1088
return b.bzrdir.create_workingtree()
1089
except errors.NotLocalUrl:
1090
# new formats - catch No tree error and create
1091
# a branch reference and a checkout.
1092
# old formats at that point - raise TestSkipped.
1093
# TODO: rbc 20060208
1094
return WorkingTreeFormat2().initialize(bzrdir.BzrDir.open(relpath))
1096
def assertIsDirectory(self, relpath, transport):
1097
"""Assert that relpath within transport is a directory.
1099
This may not be possible on all transports; in that case it propagates
1100
a TransportNotPossible.
1103
mode = transport.stat(relpath).st_mode
1104
except errors.NoSuchFile:
1105
self.fail("path %s is not a directory; no such file"
1107
if not stat.S_ISDIR(mode):
1108
self.fail("path %s is not a directory; has mode %#o"
1112
class ChrootedTestCase(TestCaseWithTransport):
1113
"""A support class that provides readonly urls outside the local namespace.
1115
This is done by checking if self.transport_server is a MemoryServer. if it
1116
is then we are chrooted already, if it is not then an HttpServer is used
1119
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
1120
be used without needed to redo it when a different
1121
subclass is in use ?
1125
super(ChrootedTestCase, self).setUp()
1126
if not self.transport_server == bzrlib.transport.memory.MemoryServer:
1127
self.transport_readonly_server = bzrlib.transport.http.HttpServer
1130
def filter_suite_by_re(suite, pattern):
1131
result = TestSuite()
1132
filter_re = re.compile(pattern)
1133
for test in iter_suite_tests(suite):
1134
if filter_re.search(test.id()):
1135
result.addTest(test)
1139
def run_suite(suite, name='test', verbose=False, pattern=".*",
1140
stop_on_failure=False, keep_output=False,
1141
transport=None, lsprof_timed=None):
1142
TestCaseInTempDir._TEST_NAME = name
1143
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
1149
pb = progress.ProgressBar()
1150
runner = TextTestRunner(stream=sys.stdout,
1152
verbosity=verbosity,
1153
keep_output=keep_output,
1155
runner.stop_on_failure=stop_on_failure
1157
suite = filter_suite_by_re(suite, pattern)
1158
result = runner.run(suite)
1159
return result.wasSuccessful()
1162
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
1165
test_suite_factory=None,
1167
"""Run the whole test suite under the enhanced runner"""
1168
global default_transport
1169
if transport is None:
1170
transport = default_transport
1171
old_transport = default_transport
1172
default_transport = transport
1174
if test_suite_factory is None:
1175
suite = test_suite()
1177
suite = test_suite_factory()
1178
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
1179
stop_on_failure=stop_on_failure, keep_output=keep_output,
1180
transport=transport,
1181
lsprof_timed=lsprof_timed)
1183
default_transport = old_transport
1187
"""Build and return TestSuite for the whole of bzrlib.
1189
This function can be replaced if you need to change the default test
1190
suite on a global basis, but it is not encouraged.
1192
from doctest import DocTestSuite
1194
global MODULES_TO_DOCTEST
1197
'bzrlib.tests.test_ancestry',
1198
'bzrlib.tests.test_api',
1199
'bzrlib.tests.test_bad_files',
1200
'bzrlib.tests.test_branch',
1201
'bzrlib.tests.test_bundle',
1202
'bzrlib.tests.test_bzrdir',
1203
'bzrlib.tests.test_command',
1204
'bzrlib.tests.test_commit',
1205
'bzrlib.tests.test_commit_merge',
1206
'bzrlib.tests.test_config',
1207
'bzrlib.tests.test_conflicts',
1208
'bzrlib.tests.test_decorators',
1209
'bzrlib.tests.test_diff',
1210
'bzrlib.tests.test_doc_generate',
1211
'bzrlib.tests.test_emptytree',
1212
'bzrlib.tests.test_errors',
1213
'bzrlib.tests.test_escaped_store',
1214
'bzrlib.tests.test_fetch',
1215
'bzrlib.tests.test_gpg',
1216
'bzrlib.tests.test_graph',
1217
'bzrlib.tests.test_hashcache',
1218
'bzrlib.tests.test_http',
1219
'bzrlib.tests.test_identitymap',
1220
'bzrlib.tests.test_inv',
1221
'bzrlib.tests.test_knit',
1222
'bzrlib.tests.test_lockdir',
1223
'bzrlib.tests.test_lockable_files',
1224
'bzrlib.tests.test_log',
1225
'bzrlib.tests.test_merge',
1226
'bzrlib.tests.test_merge3',
1227
'bzrlib.tests.test_merge_core',
1228
'bzrlib.tests.test_missing',
1229
'bzrlib.tests.test_msgeditor',
1230
'bzrlib.tests.test_nonascii',
1231
'bzrlib.tests.test_options',
1232
'bzrlib.tests.test_osutils',
1233
'bzrlib.tests.test_patch',
1234
'bzrlib.tests.test_patches',
1235
'bzrlib.tests.test_permissions',
1236
'bzrlib.tests.test_plugins',
1237
'bzrlib.tests.test_progress',
1238
'bzrlib.tests.test_reconcile',
1239
'bzrlib.tests.test_repository',
1240
'bzrlib.tests.test_revision',
1241
'bzrlib.tests.test_revisionnamespaces',
1242
'bzrlib.tests.test_revprops',
1243
'bzrlib.tests.test_revisiontree',
1244
'bzrlib.tests.test_rio',
1245
'bzrlib.tests.test_sampler',
1246
'bzrlib.tests.test_selftest',
1247
'bzrlib.tests.test_setup',
1248
'bzrlib.tests.test_sftp_transport',
1249
'bzrlib.tests.test_smart_add',
1250
'bzrlib.tests.test_source',
1251
'bzrlib.tests.test_status',
1252
'bzrlib.tests.test_store',
1253
'bzrlib.tests.test_symbol_versioning',
1254
'bzrlib.tests.test_testament',
1255
'bzrlib.tests.test_textfile',
1256
'bzrlib.tests.test_textmerge',
1257
'bzrlib.tests.test_trace',
1258
'bzrlib.tests.test_transactions',
1259
'bzrlib.tests.test_transform',
1260
'bzrlib.tests.test_transport',
1261
'bzrlib.tests.test_tsort',
1262
'bzrlib.tests.test_tuned_gzip',
1263
'bzrlib.tests.test_ui',
1264
'bzrlib.tests.test_upgrade',
1265
'bzrlib.tests.test_urlutils',
1266
'bzrlib.tests.test_versionedfile',
1267
'bzrlib.tests.test_weave',
1268
'bzrlib.tests.test_whitebox',
1269
'bzrlib.tests.test_workingtree',
1270
'bzrlib.tests.test_xml',
1272
test_transport_implementations = [
1273
'bzrlib.tests.test_transport_implementations',
1274
'bzrlib.tests.test_read_bundle',
1278
loader = TestUtil.TestLoader()
1279
from bzrlib.transport import TransportTestProviderAdapter
1280
adapter = TransportTestProviderAdapter()
1281
adapt_modules(test_transport_implementations, adapter, loader, suite)
1282
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1283
for package in packages_to_test():
1284
suite.addTest(package.test_suite())
1285
for m in MODULES_TO_TEST:
1286
suite.addTest(loader.loadTestsFromModule(m))
1287
for m in (MODULES_TO_DOCTEST):
1288
suite.addTest(DocTestSuite(m))
1289
for name, plugin in bzrlib.plugin.all_plugins().items():
1290
if getattr(plugin, 'test_suite', None) is not None:
1291
suite.addTest(plugin.test_suite())
1295
def adapt_modules(mods_list, adapter, loader, suite):
1296
"""Adapt the modules in mods_list using adapter and add to suite."""
1297
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1298
suite.addTests(adapter.adapt(test))
added
removed
bzrlib/tests/__init__.py
