15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
20
from unittest import TestLoader
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.setFormatter(bzrlib.trace.QuietFormatter())
680
handler.setLevel(logging.INFO)
681
logger = logging.getLogger('')
682
logger.addHandler(handler)
683
old_ui_factory = bzrlib.ui.ui_factory
684
bzrlib.ui.ui_factory = bzrlib.tests.blackbox.TestUIFactory(
687
bzrlib.ui.ui_factory.stdin = stdin
689
result = self.apply_redirected(stdin, stdout, stderr,
690
bzrlib.commands.run_bzr_catch_errors,
693
logger.removeHandler(handler)
694
bzrlib.ui.ui_factory = old_ui_factory
696
out = stdout.getvalue()
697
err = stderr.getvalue()
699
self.log('output:\n%r', out)
701
self.log('errors:\n%r', err)
702
if retcode is not None:
703
self.assertEquals(retcode, result)
706
def run_bzr(self, *args, **kwargs):
707
"""Invoke bzr, as if it were run from the command line.
709
This should be the main method for tests that want to exercise the
710
overall behavior of the bzr application (rather than a unit test
711
or a functional test of the library.)
713
This sends the stdout/stderr results into the test's log,
714
where it may be useful for debugging. See also run_captured.
716
:param stdin: A string to be used as stdin for the command.
718
retcode = kwargs.pop('retcode', 0)
719
encoding = kwargs.pop('encoding', None)
720
stdin = kwargs.pop('stdin', None)
721
return self.run_bzr_captured(args, retcode=retcode, encoding=encoding, stdin=stdin)
723
def run_bzr_decode(self, *args, **kwargs):
724
if kwargs.has_key('encoding'):
725
encoding = kwargs['encoding']
727
encoding = bzrlib.user_encoding
728
return self.run_bzr(*args, **kwargs)[0].decode(encoding)
730
def run_bzr_subprocess(self, *args, **kwargs):
731
"""Run bzr in a subprocess for testing.
733
This starts a new Python interpreter and runs bzr in there.
734
This should only be used for tests that have a justifiable need for
735
this isolation: e.g. they are testing startup time, or signal
736
handling, or early startup code, etc. Subprocess code can't be
737
profiled or debugged so easily.
739
:param retcode: The status code that is expected. Defaults to 0. If
740
None is supplied, the status code is not checked.
742
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
744
process = Popen([sys.executable, bzr_path]+args, stdout=PIPE,
746
out = process.stdout.read()
747
err = process.stderr.read()
748
retcode = process.wait()
749
supplied_retcode = kwargs.get('retcode', 0)
750
if supplied_retcode is not None:
751
assert supplied_retcode == retcode
754
def check_inventory_shape(self, inv, shape):
755
"""Compare an inventory to a list of expected names.
757
Fail if they are not precisely equal.
760
shape = list(shape) # copy
761
for path, ie in inv.entries():
762
name = path.replace('\\', '/')
770
self.fail("expected paths not found in inventory: %r" % shape)
772
self.fail("unexpected paths found in inventory: %r" % extras)
774
def apply_redirected(self, stdin=None, stdout=None, stderr=None,
775
a_callable=None, *args, **kwargs):
776
"""Call callable with redirected std io pipes.
778
Returns the return code."""
779
if not callable(a_callable):
780
raise ValueError("a_callable must be callable.")
784
if getattr(self, "_log_file", None) is not None:
785
stdout = self._log_file
789
if getattr(self, "_log_file", None is not None):
790
stderr = self._log_file
793
real_stdin = sys.stdin
794
real_stdout = sys.stdout
795
real_stderr = sys.stderr
800
return a_callable(*args, **kwargs)
802
sys.stdout = real_stdout
803
sys.stderr = real_stderr
804
sys.stdin = real_stdin
806
def merge(self, branch_from, wt_to):
807
"""A helper for tests to do a ui-less merge.
809
This should move to the main library when someone has time to integrate
812
# minimal ui-less merge.
813
wt_to.branch.fetch(branch_from)
814
base_rev = common_ancestor(branch_from.last_revision(),
815
wt_to.branch.last_revision(),
816
wt_to.branch.repository)
817
merge_inner(wt_to.branch, branch_from.basis_tree(),
818
wt_to.branch.repository.revision_tree(base_rev),
820
wt_to.add_pending_merge(branch_from.last_revision())
823
BzrTestBase = TestCase
826
class TestCaseInTempDir(TestCase):
827
"""Derived class that runs a test within a temporary directory.
829
This is useful for tests that need to create a branch, etc.
831
The directory is created in a slightly complex way: for each
832
Python invocation, a new temporary top-level directory is created.
833
All test cases create their own directory within that. If the
834
tests complete successfully, the directory is removed.
836
InTempDir is an old alias for FunctionalTestCase.
841
OVERRIDE_PYTHON = 'python'
843
def check_file_contents(self, filename, expect):
844
self.log("check contents of file %s" % filename)
845
contents = file(filename, 'r').read()
846
if contents != expect:
847
self.log("expected: %r" % expect)
848
self.log("actually: %r" % contents)
849
self.fail("contents of %s not as expected" % filename)
851
def _make_test_root(self):
852
if TestCaseInTempDir.TEST_ROOT is not None:
856
root = u'test%04d.tmp' % i
860
if e.errno == errno.EEXIST:
865
# successfully created
866
TestCaseInTempDir.TEST_ROOT = osutils.abspath(root)
868
# make a fake bzr directory there to prevent any tests propagating
869
# up onto the source directory's real branch
870
bzrdir.BzrDir.create_standalone_workingtree(TestCaseInTempDir.TEST_ROOT)
873
super(TestCaseInTempDir, self).setUp()
874
self._make_test_root()
875
_currentdir = os.getcwdu()
876
# shorten the name, to avoid test failures due to path length
877
short_id = self.id().replace('bzrlib.tests.', '') \
878
.replace('__main__.', '')[-100:]
879
# it's possible the same test class is run several times for
880
# parameterized tests, so make sure the names don't collide.
884
candidate_dir = '%s/%s.%d' % (self.TEST_ROOT, short_id, i)
886
candidate_dir = '%s/%s' % (self.TEST_ROOT, short_id)
887
if os.path.exists(candidate_dir):
891
self.test_dir = candidate_dir
892
os.mkdir(self.test_dir)
893
os.chdir(self.test_dir)
895
os.environ['HOME'] = self.test_dir
896
os.environ['APPDATA'] = self.test_dir
897
def _leaveDirectory():
898
os.chdir(_currentdir)
899
self.addCleanup(_leaveDirectory)
901
def build_tree(self, shape, line_endings='native', transport=None):
902
"""Build a test tree according to a pattern.
904
shape is a sequence of file specifications. If the final
905
character is '/', a directory is created.
907
This doesn't add anything to a branch.
908
:param line_endings: Either 'binary' or 'native'
909
in binary mode, exact contents are written
910
in native mode, the line endings match the
911
default platform endings.
913
:param transport: A transport to write to, for building trees on
914
VFS's. If the transport is readonly or None,
915
"." is opened automatically.
917
# XXX: It's OK to just create them using forward slashes on windows?
918
if transport is None or transport.is_readonly():
919
transport = get_transport(".")
921
self.assert_(isinstance(name, basestring))
923
transport.mkdir(urlutils.escape(name[:-1]))
925
if line_endings == 'binary':
927
elif line_endings == 'native':
930
raise errors.BzrError('Invalid line ending request %r' % (line_endings,))
931
content = "contents of %s%s" % (name.encode('utf-8'), end)
932
transport.put(urlutils.escape(name), StringIO(content))
934
def build_tree_contents(self, shape):
935
build_tree_contents(shape)
937
def failUnlessExists(self, path):
938
"""Fail unless path, which may be abs or relative, exists."""
939
self.failUnless(osutils.lexists(path))
941
def failIfExists(self, path):
942
"""Fail if path, which may be abs or relative, exists."""
943
self.failIf(osutils.lexists(path))
945
def assertFileEqual(self, content, path):
946
"""Fail if path does not contain 'content'."""
947
self.failUnless(osutils.lexists(path))
948
# TODO: jam 20060427 Shouldn't this be 'rb'?
949
self.assertEqualDiff(content, open(path, 'r').read())
952
class TestCaseWithTransport(TestCaseInTempDir):
953
"""A test case that provides get_url and get_readonly_url facilities.
955
These back onto two transport servers, one for readonly access and one for
958
If no explicit class is provided for readonly access, a
959
ReadonlyTransportDecorator is used instead which allows the use of non disk
960
based read write transports.
962
If an explicit class is provided for readonly access, that server and the
963
readwrite one must both define get_url() as resolving to os.getcwd().
966
def __init__(self, methodName='testMethod'):
967
super(TestCaseWithTransport, self).__init__(methodName)
968
self.__readonly_server = None
970
self.transport_server = default_transport
971
self.transport_readonly_server = None
973
def get_readonly_url(self, relpath=None):
974
"""Get a URL for the readonly transport.
976
This will either be backed by '.' or a decorator to the transport
977
used by self.get_url()
978
relpath provides for clients to get a path relative to the base url.
979
These should only be downwards relative, not upwards.
981
base = self.get_readonly_server().get_url()
982
if relpath is not None:
983
if not base.endswith('/'):
985
base = base + relpath
988
def get_readonly_server(self):
989
"""Get the server instance for the readonly transport
991
This is useful for some tests with specific servers to do diagnostics.
993
if self.__readonly_server is None:
994
if self.transport_readonly_server is None:
995
# readonly decorator requested
996
# bring up the server
998
self.__readonly_server = ReadonlyServer()
999
self.__readonly_server.setUp(self.__server)
1001
self.__readonly_server = self.transport_readonly_server()
1002
self.__readonly_server.setUp()
1003
self.addCleanup(self.__readonly_server.tearDown)
1004
return self.__readonly_server
1006
def get_server(self):
1007
"""Get the read/write server instance.
1009
This is useful for some tests with specific servers that need
1012
if self.__server is None:
1013
self.__server = self.transport_server()
1014
self.__server.setUp()
1015
self.addCleanup(self.__server.tearDown)
1016
return self.__server
1018
def get_url(self, relpath=None):
1019
"""Get a URL for the readwrite transport.
1021
This will either be backed by '.' or to an equivalent non-file based
1023
relpath provides for clients to get a path relative to the base url.
1024
These should only be downwards relative, not upwards.
1026
base = self.get_server().get_url()
1027
if relpath is not None and relpath != '.':
1028
if not base.endswith('/'):
1030
base = base + urlutils.escape(relpath)
1033
def get_transport(self):
1034
"""Return a writeable transport for the test scratch space"""
1035
t = get_transport(self.get_url())
1036
self.assertFalse(t.is_readonly())
1039
def get_readonly_transport(self):
1040
"""Return a readonly transport for the test scratch space
1042
This can be used to test that operations which should only need
1043
readonly access in fact do not try to write.
1045
t = get_transport(self.get_readonly_url())
1046
self.assertTrue(t.is_readonly())
1049
def make_branch(self, relpath, format=None):
1050
"""Create a branch on the transport at relpath."""
1051
repo = self.make_repository(relpath, format=format)
1052
return repo.bzrdir.create_branch()
1054
def make_bzrdir(self, relpath, format=None):
1056
url = self.get_url(relpath)
1057
mutter('relpath %r => url %r', relpath, url)
1058
segments = url.split('/')
1059
if segments and segments[-1] not in ('', '.'):
1060
parent = '/'.join(segments[:-1])
1061
t = get_transport(parent)
1063
t.mkdir(segments[-1])
1064
except errors.FileExists:
1067
format=bzrlib.bzrdir.BzrDirFormat.get_default_format()
1068
# FIXME: make this use a single transport someday. RBC 20060418
1069
return format.initialize_on_transport(get_transport(relpath))
1070
except errors.UninitializableFormat:
1071
raise TestSkipped("Format %s is not initializable." % format)
1073
def make_repository(self, relpath, shared=False, format=None):
1074
"""Create a repository on our default transport at relpath."""
1075
made_control = self.make_bzrdir(relpath, format=format)
1076
return made_control.create_repository(shared=shared)
1078
def make_branch_and_tree(self, relpath, format=None):
1079
"""Create a branch on the transport and a tree locally.
1083
# TODO: always use the local disk path for the working tree,
1084
# this obviously requires a format that supports branch references
1085
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
1087
b = self.make_branch(relpath, format=format)
1089
return b.bzrdir.create_workingtree()
1090
except errors.NotLocalUrl:
1091
# new formats - catch No tree error and create
1092
# a branch reference and a checkout.
1093
# old formats at that point - raise TestSkipped.
1094
# TODO: rbc 20060208
1095
return WorkingTreeFormat2().initialize(bzrdir.BzrDir.open(relpath))
1097
def assertIsDirectory(self, relpath, transport):
1098
"""Assert that relpath within transport is a directory.
1100
This may not be possible on all transports; in that case it propagates
1101
a TransportNotPossible.
1104
mode = transport.stat(relpath).st_mode
1105
except errors.NoSuchFile:
1106
self.fail("path %s is not a directory; no such file"
1108
if not stat.S_ISDIR(mode):
1109
self.fail("path %s is not a directory; has mode %#o"
1113
class ChrootedTestCase(TestCaseWithTransport):
1114
"""A support class that provides readonly urls outside the local namespace.
1116
This is done by checking if self.transport_server is a MemoryServer. if it
1117
is then we are chrooted already, if it is not then an HttpServer is used
1120
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
1121
be used without needed to redo it when a different
1122
subclass is in use ?
1126
super(ChrootedTestCase, self).setUp()
1127
if not self.transport_server == bzrlib.transport.memory.MemoryServer:
1128
self.transport_readonly_server = bzrlib.transport.http.HttpServer
1131
def filter_suite_by_re(suite, pattern):
1132
result = TestSuite()
1133
filter_re = re.compile(pattern)
1134
for test in iter_suite_tests(suite):
1135
if filter_re.search(test.id()):
1136
result.addTest(test)
1140
def run_suite(suite, name='test', verbose=False, pattern=".*",
1141
stop_on_failure=False, keep_output=False,
1142
transport=None, lsprof_timed=None):
1143
TestCaseInTempDir._TEST_NAME = name
1144
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
1150
pb = progress.ProgressBar()
1151
runner = TextTestRunner(stream=sys.stdout,
1153
verbosity=verbosity,
1154
keep_output=keep_output,
1156
runner.stop_on_failure=stop_on_failure
1158
suite = filter_suite_by_re(suite, pattern)
1159
result = runner.run(suite)
1160
return result.wasSuccessful()
1163
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
1166
test_suite_factory=None,
1168
"""Run the whole test suite under the enhanced runner"""
1169
global default_transport
1170
if transport is None:
1171
transport = default_transport
1172
old_transport = default_transport
1173
default_transport = transport
1175
if test_suite_factory is None:
1176
suite = test_suite()
1178
suite = test_suite_factory()
1179
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
1180
stop_on_failure=stop_on_failure, keep_output=keep_output,
1181
transport=transport,
1182
lsprof_timed=lsprof_timed)
1184
default_transport = old_transport
1188
"""Build and return TestSuite for the whole of bzrlib.
1190
This function can be replaced if you need to change the default test
1191
suite on a global basis, but it is not encouraged.
22
1193
from doctest import DocTestSuite
24
tr = unittest.TextTestRunner(verbosity=2)
25
suite = unittest.TestSuite()
26
import bzrlib.whitebox
28
suite.addTest(TestLoader().loadTestsFromModule(bzrlib.whitebox))
30
for m in bzrlib.store, bzrlib.inventory, bzrlib.branch, bzrlib.osutils, \
1195
global MODULES_TO_DOCTEST
1198
'bzrlib.tests.test_ancestry',
1199
'bzrlib.tests.test_api',
1200
'bzrlib.tests.test_bad_files',
1201
'bzrlib.tests.test_branch',
1202
'bzrlib.tests.test_bundle',
1203
'bzrlib.tests.test_bzrdir',
1204
'bzrlib.tests.test_command',
1205
'bzrlib.tests.test_commit',
1206
'bzrlib.tests.test_commit_merge',
1207
'bzrlib.tests.test_config',
1208
'bzrlib.tests.test_conflicts',
1209
'bzrlib.tests.test_decorators',
1210
'bzrlib.tests.test_diff',
1211
'bzrlib.tests.test_doc_generate',
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_rio',
1244
'bzrlib.tests.test_sampler',
1245
'bzrlib.tests.test_selftest',
1246
'bzrlib.tests.test_setup',
1247
'bzrlib.tests.test_sftp_transport',
1248
'bzrlib.tests.test_smart_add',
1249
'bzrlib.tests.test_source',
1250
'bzrlib.tests.test_status',
1251
'bzrlib.tests.test_store',
1252
'bzrlib.tests.test_symbol_versioning',
1253
'bzrlib.tests.test_testament',
1254
'bzrlib.tests.test_textfile',
1255
'bzrlib.tests.test_textmerge',
1256
'bzrlib.tests.test_trace',
1257
'bzrlib.tests.test_transactions',
1258
'bzrlib.tests.test_transform',
1259
'bzrlib.tests.test_transport',
1260
'bzrlib.tests.test_tsort',
1261
'bzrlib.tests.test_tuned_gzip',
1262
'bzrlib.tests.test_ui',
1263
'bzrlib.tests.test_upgrade',
1264
'bzrlib.tests.test_urlutils',
1265
'bzrlib.tests.test_versionedfile',
1266
'bzrlib.tests.test_weave',
1267
'bzrlib.tests.test_whitebox',
1268
'bzrlib.tests.test_workingtree',
1269
'bzrlib.tests.test_xml',
1271
test_transport_implementations = [
1272
'bzrlib.tests.test_transport_implementations']
1275
loader = TestUtil.TestLoader()
1276
from bzrlib.transport import TransportTestProviderAdapter
1277
adapter = TransportTestProviderAdapter()
1278
adapt_modules(test_transport_implementations, adapter, loader, suite)
1279
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
1280
for package in packages_to_test():
1281
suite.addTest(package.test_suite())
1282
for m in MODULES_TO_TEST:
1283
suite.addTest(loader.loadTestsFromModule(m))
1284
for m in (MODULES_TO_DOCTEST):
32
1285
suite.addTest(DocTestSuite(m))
34
result = tr.run(suite)
35
return result.wasSuccessful()
1286
for name, plugin in bzrlib.plugin.all_plugins().items():
1287
if getattr(plugin, 'test_suite', None) is not None:
1288
suite.addTest(plugin.test_suite())
1292
def adapt_modules(mods_list, adapter, loader, suite):
1293
"""Adapt the modules in mods_list using adapter and add to suite."""
1294
for test in iter_suite_tests(loader.loadTestsFromModuleNames(mods_list)):
1295
suite.addTests(adapter.adapt(test))
added
removed
bzrlib/tests/__init__.py
