1
# Copyright (C) 2005, 2006, 2007, 2008 Canonical Ltd
1
# Copyright (C) 2005 by Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
5
5
# the Free Software Foundation; either version 2 of the License, or
6
6
# (at your option) any later version.
8
8
# This program is distributed in the hope that it will be useful,
9
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
11
11
# GNU General Public License for more details.
13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
18
# TODO: Perhaps there should be an API to find out if bzr running under the
19
# test suite -- some plugins might want to avoid making intrusive changes if
20
# this is the case. However, we want behaviour under to test to diverge as
21
# little as possible, so this should be used rarely if it's added at all.
22
# (Suggestion from j-a-meinel, 2005-11-24)
24
# NOTE: Some classes in here use camelCaseNaming() rather than
25
# underscore_naming(). That's for consistency with unittest; it's not the
26
# general style of bzrlib. Please continue that consistency when adding e.g.
27
# new assertFoo() methods.
31
18
from cStringIO import StringIO
37
from pprint import pformat
42
from subprocess import Popen, PIPE
63
29
import bzrlib.commands
64
import bzrlib.timestamp
66
import bzrlib.inventory
67
import bzrlib.iterablefile
72
# lsprof not available
74
from bzrlib.merge import merge_inner
78
from bzrlib import symbol_versioning
79
from bzrlib.symbol_versioning import (
85
30
import bzrlib.trace
86
from bzrlib.transport import get_transport
87
import bzrlib.transport
88
from bzrlib.transport.local import LocalURLServer
89
from bzrlib.transport.memory import MemoryServer
90
from bzrlib.transport.readonly import ReadonlyServer
91
from bzrlib.trace import mutter, note
92
from bzrlib.tests import TestUtil
93
from bzrlib.tests.http_server import HttpServer
94
from bzrlib.tests.TestUtil import (
98
from bzrlib.tests.treeshape import build_tree_contents
99
import bzrlib.version_info_formats.format_custom
100
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
102
# Mark this python module as being part of the implementation
103
# of unittest: this gives us better tracebacks where the last
104
# shown frame is the test code, not our assertXYZ.
107
default_transport = LocalURLServer
110
class ExtendedTestResult(unittest._TextTestResult):
111
"""Accepts, reports and accumulates the results of running tests.
113
Compared to the unittest version this class adds support for
114
profiling, benchmarking, stopping as soon as a test fails, and
115
skipping tests. There are further-specialized subclasses for
116
different types of display.
118
When a test finishes, in whatever way, it calls one of the addSuccess,
119
addFailure or addError classes. These in turn may redirect to a more
120
specific case for the special test results supported by our extended
123
Note that just one of these objects is fed the results from many tests.
128
def __init__(self, stream, descriptions, verbosity,
132
"""Construct new TestResult.
134
:param bench_history: Optionally, a writable file object to accumulate
137
unittest._TextTestResult.__init__(self, stream, descriptions, verbosity)
138
if bench_history is not None:
139
from bzrlib.version import _get_bzr_source_tree
140
src_tree = _get_bzr_source_tree()
143
revision_id = src_tree.get_parent_ids()[0]
145
# XXX: if this is a brand new tree, do the same as if there
149
# XXX: If there's no branch, what should we do?
151
bench_history.write("--date %s %s\n" % (time.time(), revision_id))
152
self._bench_history = bench_history
153
self.ui = ui.ui_factory
154
self.num_tests = num_tests
156
self.failure_count = 0
157
self.known_failure_count = 0
159
self.not_applicable_count = 0
160
self.unsupported = {}
162
self._overall_start_time = time.time()
164
def _extractBenchmarkTime(self, testCase):
165
"""Add a benchmark time for the current test case."""
166
return getattr(testCase, "_benchtime", None)
168
def _elapsedTestTimeString(self):
169
"""Return a time string for the overall time the current test has taken."""
170
return self._formatTime(time.time() - self._start_time)
172
def _testTimeString(self, testCase):
173
benchmark_time = self._extractBenchmarkTime(testCase)
174
if benchmark_time is not None:
176
self._formatTime(benchmark_time),
177
self._elapsedTestTimeString())
179
return " %s" % self._elapsedTestTimeString()
181
def _formatTime(self, seconds):
182
"""Format seconds as milliseconds with leading spaces."""
183
# some benchmarks can take thousands of seconds to run, so we need 8
185
return "%8dms" % (1000 * seconds)
187
def _shortened_test_description(self, test):
189
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
192
def startTest(self, test):
193
unittest.TestResult.startTest(self, test)
194
self.report_test_start(test)
195
test.number = self.count
196
self._recordTestStartTime()
198
def _recordTestStartTime(self):
199
"""Record that a test has started."""
200
self._start_time = time.time()
202
def _cleanupLogFile(self, test):
203
# We can only do this if we have one of our TestCases, not if
205
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
206
if setKeepLogfile is not None:
209
def addError(self, test, err):
210
"""Tell result that test finished with an error.
212
Called from the TestCase run() method when the test
213
fails with an unexpected error.
215
self._testConcluded(test)
216
if isinstance(err[1], TestSkipped):
217
return self._addSkipped(test, err)
218
elif isinstance(err[1], UnavailableFeature):
219
return self.addNotSupported(test, err[1].args[0])
221
self._cleanupLogFile(test)
222
unittest.TestResult.addError(self, test, err)
223
self.error_count += 1
224
self.report_error(test, err)
228
def addFailure(self, test, err):
229
"""Tell result that test failed.
231
Called from the TestCase run() method when the test
232
fails because e.g. an assert() method failed.
234
self._testConcluded(test)
235
if isinstance(err[1], KnownFailure):
236
return self._addKnownFailure(test, err)
238
self._cleanupLogFile(test)
239
unittest.TestResult.addFailure(self, test, err)
240
self.failure_count += 1
241
self.report_failure(test, err)
245
def addSuccess(self, test):
246
"""Tell result that test completed successfully.
248
Called from the TestCase run()
250
self._testConcluded(test)
251
if self._bench_history is not None:
252
benchmark_time = self._extractBenchmarkTime(test)
253
if benchmark_time is not None:
254
self._bench_history.write("%s %s\n" % (
255
self._formatTime(benchmark_time),
257
self.report_success(test)
258
self._cleanupLogFile(test)
259
unittest.TestResult.addSuccess(self, test)
260
test._log_contents = ''
262
def _testConcluded(self, test):
263
"""Common code when a test has finished.
265
Called regardless of whether it succeded, failed, etc.
269
def _addKnownFailure(self, test, err):
270
self.known_failure_count += 1
271
self.report_known_failure(test, err)
273
def addNotSupported(self, test, feature):
274
"""The test will not be run because of a missing feature.
276
# this can be called in two different ways: it may be that the
277
# test started running, and then raised (through addError)
278
# UnavailableFeature. Alternatively this method can be called
279
# while probing for features before running the tests; in that
280
# case we will see startTest and stopTest, but the test will never
282
self.unsupported.setdefault(str(feature), 0)
283
self.unsupported[str(feature)] += 1
284
self.report_unsupported(test, feature)
286
def _addSkipped(self, test, skip_excinfo):
287
if isinstance(skip_excinfo[1], TestNotApplicable):
288
self.not_applicable_count += 1
289
self.report_not_applicable(test, skip_excinfo)
292
self.report_skip(test, skip_excinfo)
295
except KeyboardInterrupt:
298
self.addError(test, test._exc_info())
300
# seems best to treat this as success from point-of-view of unittest
301
# -- it actually does nothing so it barely matters :)
302
unittest.TestResult.addSuccess(self, test)
303
test._log_contents = ''
305
def printErrorList(self, flavour, errors):
306
for test, err in errors:
307
self.stream.writeln(self.separator1)
308
self.stream.write("%s: " % flavour)
309
self.stream.writeln(self.getDescription(test))
310
if getattr(test, '_get_log', None) is not None:
311
self.stream.write('\n')
313
('vvvv[log from %s]' % test.id()).ljust(78,'-'))
314
self.stream.write('\n')
315
self.stream.write(test._get_log())
316
self.stream.write('\n')
318
('^^^^[log from %s]' % test.id()).ljust(78,'-'))
319
self.stream.write('\n')
320
self.stream.writeln(self.separator2)
321
self.stream.writeln("%s" % err)
326
def report_cleaning_up(self):
329
def report_success(self, test):
332
def wasStrictlySuccessful(self):
333
if self.unsupported or self.known_failure_count:
335
return self.wasSuccessful()
338
class TextTestResult(ExtendedTestResult):
339
"""Displays progress and results of tests in text form"""
341
def __init__(self, stream, descriptions, verbosity,
346
ExtendedTestResult.__init__(self, stream, descriptions, verbosity,
347
bench_history, num_tests)
349
self.pb = self.ui.nested_progress_bar()
350
self._supplied_pb = False
353
self._supplied_pb = True
354
self.pb.show_pct = False
355
self.pb.show_spinner = False
356
self.pb.show_eta = False,
357
self.pb.show_count = False
358
self.pb.show_bar = False
360
def report_starting(self):
361
self.pb.update('[test 0/%d] starting...' % (self.num_tests))
363
def _progress_prefix_text(self):
364
# the longer this text, the less space we have to show the test
366
a = '[%d' % self.count # total that have been run
367
# tests skipped as known not to be relevant are not important enough
369
## if self.skip_count:
370
## a += ', %d skip' % self.skip_count
371
## if self.known_failure_count:
372
## a += '+%dX' % self.known_failure_count
373
if self.num_tests is not None:
374
a +='/%d' % self.num_tests
376
runtime = time.time() - self._overall_start_time
378
a += '%dm%ds' % (runtime / 60, runtime % 60)
382
a += ', %d err' % self.error_count
383
if self.failure_count:
384
a += ', %d fail' % self.failure_count
386
a += ', %d missing' % len(self.unsupported)
390
def report_test_start(self, test):
393
self._progress_prefix_text()
395
+ self._shortened_test_description(test))
397
def _test_description(self, test):
398
return self._shortened_test_description(test)
400
def report_error(self, test, err):
401
self.pb.note('ERROR: %s\n %s\n',
402
self._test_description(test),
406
def report_failure(self, test, err):
407
self.pb.note('FAIL: %s\n %s\n',
408
self._test_description(test),
412
def report_known_failure(self, test, err):
413
self.pb.note('XFAIL: %s\n%s\n',
414
self._test_description(test), err[1])
416
def report_skip(self, test, skip_excinfo):
419
def report_not_applicable(self, test, skip_excinfo):
422
def report_unsupported(self, test, feature):
423
"""test cannot be run because feature is missing."""
425
def report_cleaning_up(self):
426
self.pb.update('cleaning up...')
429
if not self._supplied_pb:
433
class VerboseTestResult(ExtendedTestResult):
434
"""Produce long output, with one line per test run plus times"""
436
def _ellipsize_to_right(self, a_string, final_width):
437
"""Truncate and pad a string, keeping the right hand side"""
438
if len(a_string) > final_width:
439
result = '...' + a_string[3-final_width:]
442
return result.ljust(final_width)
444
def report_starting(self):
445
self.stream.write('running %d tests...\n' % self.num_tests)
447
def report_test_start(self, test):
449
name = self._shortened_test_description(test)
450
# width needs space for 6 char status, plus 1 for slash, plus 2 10-char
451
# numbers, plus a trailing blank
452
# when NUMBERED_DIRS: plus 5 chars on test number, plus 1 char on space
453
self.stream.write(self._ellipsize_to_right(name,
454
osutils.terminal_width()-30))
457
def _error_summary(self, err):
459
return '%s%s' % (indent, err[1])
461
def report_error(self, test, err):
462
self.stream.writeln('ERROR %s\n%s'
463
% (self._testTimeString(test),
464
self._error_summary(err)))
466
def report_failure(self, test, err):
467
self.stream.writeln(' FAIL %s\n%s'
468
% (self._testTimeString(test),
469
self._error_summary(err)))
471
def report_known_failure(self, test, err):
472
self.stream.writeln('XFAIL %s\n%s'
473
% (self._testTimeString(test),
474
self._error_summary(err)))
476
def report_success(self, test):
477
self.stream.writeln(' OK %s' % self._testTimeString(test))
478
for bench_called, stats in getattr(test, '_benchcalls', []):
479
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
480
stats.pprint(file=self.stream)
481
# flush the stream so that we get smooth output. This verbose mode is
482
# used to show the output in PQM.
485
def report_skip(self, test, skip_excinfo):
486
self.stream.writeln(' SKIP %s\n%s'
487
% (self._testTimeString(test),
488
self._error_summary(skip_excinfo)))
490
def report_not_applicable(self, test, skip_excinfo):
491
self.stream.writeln(' N/A %s\n%s'
492
% (self._testTimeString(test),
493
self._error_summary(skip_excinfo)))
495
def report_unsupported(self, test, feature):
496
"""test cannot be run because feature is missing."""
497
self.stream.writeln("NODEP %s\n The feature '%s' is not available."
498
%(self._testTimeString(test), feature))
501
class TextTestRunner(object):
502
stop_on_failure = False
511
self.stream = unittest._WritelnDecorator(stream)
512
self.descriptions = descriptions
513
self.verbosity = verbosity
514
self._bench_history = bench_history
515
self.list_only = list_only
518
"Run the given test case or test suite."
519
startTime = time.time()
520
if self.verbosity == 1:
521
result_class = TextTestResult
522
elif self.verbosity >= 2:
523
result_class = VerboseTestResult
524
result = result_class(self.stream,
527
bench_history=self._bench_history,
528
num_tests=test.countTestCases(),
530
result.stop_early = self.stop_on_failure
531
result.report_starting()
533
if self.verbosity >= 2:
534
self.stream.writeln("Listing tests only ...\n")
536
for t in iter_suite_tests(test):
537
self.stream.writeln("%s" % (t.id()))
539
actionTaken = "Listed"
542
run = result.testsRun
544
stopTime = time.time()
545
timeTaken = stopTime - startTime
547
self.stream.writeln(result.separator2)
548
self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
549
run, run != 1 and "s" or "", timeTaken))
550
self.stream.writeln()
551
if not result.wasSuccessful():
552
self.stream.write("FAILED (")
553
failed, errored = map(len, (result.failures, result.errors))
555
self.stream.write("failures=%d" % failed)
557
if failed: self.stream.write(", ")
558
self.stream.write("errors=%d" % errored)
559
if result.known_failure_count:
560
if failed or errored: self.stream.write(", ")
561
self.stream.write("known_failure_count=%d" %
562
result.known_failure_count)
563
self.stream.writeln(")")
565
if result.known_failure_count:
566
self.stream.writeln("OK (known_failures=%d)" %
567
result.known_failure_count)
569
self.stream.writeln("OK")
570
if result.skip_count > 0:
571
skipped = result.skip_count
572
self.stream.writeln('%d test%s skipped' %
573
(skipped, skipped != 1 and "s" or ""))
574
if result.unsupported:
575
for feature, count in sorted(result.unsupported.items()):
576
self.stream.writeln("Missing feature '%s' skipped %d tests." %
582
def iter_suite_tests(suite):
583
"""Return all tests in a suite, recursing through nested suites"""
584
for item in suite._tests:
585
if isinstance(item, unittest.TestCase):
587
elif isinstance(item, unittest.TestSuite):
588
for r in iter_suite_tests(item):
591
raise Exception('unknown object %r inside test suite %r'
595
class TestSkipped(Exception):
596
"""Indicates that a test was intentionally skipped, rather than failing."""
599
class TestNotApplicable(TestSkipped):
600
"""A test is not applicable to the situation where it was run.
602
This is only normally raised by parameterized tests, if they find that
603
the instance they're constructed upon does not support one aspect
608
class KnownFailure(AssertionError):
609
"""Indicates that a test failed in a precisely expected manner.
611
Such failures dont block the whole test suite from passing because they are
612
indicators of partially completed code or of future work. We have an
613
explicit error for them so that we can ensure that they are always visible:
614
KnownFailures are always shown in the output of bzr selftest.
618
class UnavailableFeature(Exception):
619
"""A feature required for this test was not available.
621
The feature should be used to construct the exception.
35
MODULES_TO_DOCTEST = []
37
from logging import debug, warning, error
625
39
class CommandFailed(Exception):
629
class StringIOWrapper(object):
630
"""A wrapper around cStringIO which just adds an encoding attribute.
632
Internally we can check sys.stdout to see what the output encoding
633
should be. However, cStringIO has no encoding attribute that we can
634
set. So we wrap it instead.
639
def __init__(self, s=None):
641
self.__dict__['_cstring'] = StringIO(s)
643
self.__dict__['_cstring'] = StringIO()
645
def __getattr__(self, name, getattr=getattr):
646
return getattr(self.__dict__['_cstring'], name)
648
def __setattr__(self, name, val):
649
if name == 'encoding':
650
self.__dict__['encoding'] = val
652
return setattr(self._cstring, name, val)
655
class TestUIFactory(ui.CLIUIFactory):
656
"""A UI Factory for testing.
658
Hide the progress bar but emit note()s.
660
Allows get_password to be tested without real tty attached.
667
super(TestUIFactory, self).__init__()
668
if stdin is not None:
669
# We use a StringIOWrapper to be able to test various
670
# encodings, but the user is still responsible to
671
# encode the string and to set the encoding attribute
672
# of StringIOWrapper.
673
self.stdin = StringIOWrapper(stdin)
675
self.stdout = sys.stdout
679
self.stderr = sys.stderr
684
"""See progress.ProgressBar.clear()."""
686
def clear_term(self):
687
"""See progress.ProgressBar.clear_term()."""
689
def clear_term(self):
690
"""See progress.ProgressBar.clear_term()."""
693
"""See progress.ProgressBar.finished()."""
695
def note(self, fmt_string, *args, **kwargs):
696
"""See progress.ProgressBar.note()."""
697
self.stdout.write((fmt_string + "\n") % args)
699
def progress_bar(self):
702
def nested_progress_bar(self):
705
def update(self, message, count=None, total=None):
706
"""See progress.ProgressBar.update()."""
708
def get_non_echoed_password(self, prompt):
709
"""Get password from stdin without trying to handle the echo mode"""
711
self.stdout.write(prompt.encode(self.stdout.encoding, 'replace'))
712
password = self.stdin.readline()
715
if password[-1] == '\n':
716
password = password[:-1]
720
def _report_leaked_threads():
721
bzrlib.trace.warning('%s is leaking threads among %d leaking tests',
722
TestCase._first_thread_leaker_id,
723
TestCase._leaking_threads_tests)
726
42
class TestCase(unittest.TestCase):
727
43
"""Base class for bzr unit tests.
732
48
Error and debug log messages are redirected from their usual
733
49
location into a temporary file, the contents of which can be
734
retrieved by _get_log(). We use a real OS file, not an in-memory object,
735
so that it can also capture file IO. When the test completes this file
736
is read into memory and removed from disk.
50
retrieved by _get_log().
738
52
There are also convenience functions to invoke bzr's command-line
739
routine, and to build and check bzr trees.
741
In addition to the usual method of overriding tearDown(), this class also
742
allows subclasses to register functions into the _cleanups list, which is
743
run in order as the object is torn down. It's less likely this will be
744
accidentally overlooked.
747
_active_threads = None
748
_leaking_threads_tests = 0
749
_first_thread_leaker_id = None
750
_log_file_name = None
752
_keep_log_file = False
753
# record lsprof data when performing benchmark calls.
754
_gather_lsprof_in_benchmarks = False
755
attrs_to_keep = ('id', '_testMethodName', '_testMethodDoc',
756
'_log_contents', '_log_file_name', '_benchtime',
757
'_TestCase__testMethodName')
759
def __init__(self, methodName='testMethod'):
760
super(TestCase, self).__init__(methodName)
53
routine, and to build and check bzr trees."""
58
# this replaces the default testsweet.TestCase; we don't want logging changed
764
59
unittest.TestCase.setUp(self)
765
self._cleanEnvironment()
768
self._benchcalls = []
769
self._benchtime = None
771
self._clear_debug_flags()
772
TestCase._active_threads = threading.activeCount()
773
self.addCleanup(self._check_leaked_threads)
775
def _check_leaked_threads(self):
776
active = threading.activeCount()
777
leaked_threads = active - TestCase._active_threads
778
TestCase._active_threads = active
780
TestCase._leaking_threads_tests += 1
781
if TestCase._first_thread_leaker_id is None:
782
TestCase._first_thread_leaker_id = self.id()
783
# we're not specifically told when all tests are finished.
784
# This will do. We use a function to avoid keeping a reference
785
# to a TestCase object.
786
atexit.register(_report_leaked_threads)
788
def _clear_debug_flags(self):
789
"""Prevent externally set debug flags affecting tests.
791
Tests that want to use debug flags can just set them in the
792
debug_flags set during setup/teardown.
794
if 'allow_debug' not in selftest_debug_flags:
795
self._preserved_debug_flags = set(debug.debug_flags)
796
debug.debug_flags.clear()
797
self.addCleanup(self._restore_debug_flags)
799
def _clear_hooks(self):
800
# prevent hooks affecting tests
802
import bzrlib.smart.server
803
self._preserved_hooks = {
804
bzrlib.branch.Branch: bzrlib.branch.Branch.hooks,
805
bzrlib.mutabletree.MutableTree: bzrlib.mutabletree.MutableTree.hooks,
806
bzrlib.smart.server.SmartTCPServer: bzrlib.smart.server.SmartTCPServer.hooks,
808
self.addCleanup(self._restoreHooks)
809
# reset all hooks to an empty instance of the appropriate type
810
bzrlib.branch.Branch.hooks = bzrlib.branch.BranchHooks()
811
bzrlib.smart.server.SmartTCPServer.hooks = bzrlib.smart.server.SmartServerHooks()
813
def _silenceUI(self):
814
"""Turn off UI for duration of test"""
815
# by default the UI is off; tests can turn it on if they want it.
816
saved = ui.ui_factory
818
ui.ui_factory = saved
819
ui.ui_factory = ui.SilentUIFactory()
820
self.addCleanup(_restore)
822
def _ndiff_strings(self, a, b):
823
"""Return ndiff between two strings containing lines.
825
A trailing newline is added if missing to make the strings
827
if b and b[-1] != '\n':
829
if a and a[-1] != '\n':
831
difflines = difflib.ndiff(a.splitlines(True),
833
linejunk=lambda x: False,
834
charjunk=lambda x: False)
835
return ''.join(difflines)
837
def assertEqual(self, a, b, message=''):
841
except UnicodeError, e:
842
# If we can't compare without getting a UnicodeError, then
843
# obviously they are different
844
mutter('UnicodeError: %s', e)
847
raise AssertionError("%snot equal:\na = %s\nb = %s\n"
849
pformat(a), pformat(b)))
851
assertEquals = assertEqual
853
def assertEqualDiff(self, a, b, message=None):
854
"""Assert two texts are equal, if not raise an exception.
856
This is intended for use with multi-line strings where it can
857
be hard to find the differences by eye.
859
# TODO: perhaps override assertEquals to call this for strings?
863
message = "texts not equal:\n"
865
message = 'first string is missing a final newline.\n'
867
message = 'second string is missing a final newline.\n'
868
raise AssertionError(message +
869
self._ndiff_strings(a, b))
871
def assertEqualMode(self, mode, mode_test):
872
self.assertEqual(mode, mode_test,
873
'mode mismatch %o != %o' % (mode, mode_test))
875
def assertPositive(self, val):
876
"""Assert that val is greater than 0."""
877
self.assertTrue(val > 0, 'expected a positive value, but got %s' % val)
879
def assertNegative(self, val):
880
"""Assert that val is less than 0."""
881
self.assertTrue(val < 0, 'expected a negative value, but got %s' % val)
883
def assertStartsWith(self, s, prefix):
884
if not s.startswith(prefix):
885
raise AssertionError('string %r does not start with %r' % (s, prefix))
887
def assertEndsWith(self, s, suffix):
888
"""Asserts that s ends with suffix."""
889
if not s.endswith(suffix):
890
raise AssertionError('string %r does not end with %r' % (s, suffix))
892
def assertContainsRe(self, haystack, needle_re):
893
"""Assert that a contains something matching a regular expression."""
894
if not re.search(needle_re, haystack):
895
if '\n' in haystack or len(haystack) > 60:
896
# a long string, format it in a more readable way
897
raise AssertionError(
898
'pattern "%s" not found in\n"""\\\n%s"""\n'
899
% (needle_re, haystack))
901
raise AssertionError('pattern "%s" not found in "%s"'
902
% (needle_re, haystack))
904
def assertNotContainsRe(self, haystack, needle_re):
905
"""Assert that a does not match a regular expression"""
906
if re.search(needle_re, haystack):
907
raise AssertionError('pattern "%s" found in "%s"'
908
% (needle_re, haystack))
910
def assertSubset(self, sublist, superlist):
911
"""Assert that every entry in sublist is present in superlist."""
912
missing = set(sublist) - set(superlist)
914
raise AssertionError("value(s) %r not present in container %r" %
915
(missing, superlist))
917
def assertListRaises(self, excClass, func, *args, **kwargs):
918
"""Fail unless excClass is raised when the iterator from func is used.
920
Many functions can return generators this makes sure
921
to wrap them in a list() call to make sure the whole generator
922
is run, and that the proper exception is raised.
925
list(func(*args, **kwargs))
929
if getattr(excClass,'__name__', None) is not None:
930
excName = excClass.__name__
932
excName = str(excClass)
933
raise self.failureException, "%s not raised" % excName
935
def assertRaises(self, excClass, callableObj, *args, **kwargs):
936
"""Assert that a callable raises a particular exception.
938
:param excClass: As for the except statement, this may be either an
939
exception class, or a tuple of classes.
940
:param callableObj: A callable, will be passed ``*args`` and
943
Returns the exception so that you can examine it.
946
callableObj(*args, **kwargs)
950
if getattr(excClass,'__name__', None) is not None:
951
excName = excClass.__name__
954
excName = str(excClass)
955
raise self.failureException, "%s not raised" % excName
957
def assertIs(self, left, right, message=None):
958
if not (left is right):
959
if message is not None:
960
raise AssertionError(message)
962
raise AssertionError("%r is not %r." % (left, right))
964
def assertIsNot(self, left, right, message=None):
966
if message is not None:
967
raise AssertionError(message)
969
raise AssertionError("%r is %r." % (left, right))
971
def assertTransportMode(self, transport, path, mode):
972
"""Fail if a path does not have mode mode.
974
If modes are not supported on this transport, the assertion is ignored.
976
if not transport._can_roundtrip_unix_modebits():
978
path_stat = transport.stat(path)
979
actual_mode = stat.S_IMODE(path_stat.st_mode)
980
self.assertEqual(mode, actual_mode,
981
'mode of %r incorrect (%o != %o)' % (path, mode, actual_mode))
983
def assertIsSameRealPath(self, path1, path2):
984
"""Fail if path1 and path2 points to different files"""
985
self.assertEqual(osutils.realpath(path1),
986
osutils.realpath(path2),
987
"apparent paths:\na = %s\nb = %s\n," % (path1, path2))
989
def assertIsInstance(self, obj, kls):
990
"""Fail if obj is not an instance of kls"""
991
if not isinstance(obj, kls):
992
self.fail("%r is an instance of %s rather than %s" % (
993
obj, obj.__class__, kls))
995
def expectFailure(self, reason, assertion, *args, **kwargs):
996
"""Invoke a test, expecting it to fail for the given reason.
998
This is for assertions that ought to succeed, but currently fail.
999
(The failure is *expected* but not *wanted*.) Please be very precise
1000
about the failure you're expecting. If a new bug is introduced,
1001
AssertionError should be raised, not KnownFailure.
1003
Frequently, expectFailure should be followed by an opposite assertion.
1006
Intended to be used with a callable that raises AssertionError as the
1007
'assertion' parameter. args and kwargs are passed to the 'assertion'.
1009
Raises KnownFailure if the test fails. Raises AssertionError if the
1014
self.expectFailure('Math is broken', self.assertNotEqual, 54,
1016
self.assertEqual(42, dynamic_val)
1018
This means that a dynamic_val of 54 will cause the test to raise
1019
a KnownFailure. Once math is fixed and the expectFailure is removed,
1020
only a dynamic_val of 42 will allow the test to pass. Anything other
1021
than 54 or 42 will cause an AssertionError.
1024
assertion(*args, **kwargs)
1025
except AssertionError:
1026
raise KnownFailure(reason)
1028
self.fail('Unexpected success. Should have failed: %s' % reason)
1030
def assertFileEqual(self, content, path):
1031
"""Fail if path does not contain 'content'."""
1032
self.failUnlessExists(path)
1033
f = file(path, 'rb')
1038
self.assertEqualDiff(content, s)
1040
def failUnlessExists(self, path):
1041
"""Fail unless path or paths, which may be abs or relative, exist."""
1042
if not isinstance(path, basestring):
1044
self.failUnlessExists(p)
1046
self.failUnless(osutils.lexists(path),path+" does not exist")
1048
def failIfExists(self, path):
1049
"""Fail if path or paths, which may be abs or relative, exist."""
1050
if not isinstance(path, basestring):
1052
self.failIfExists(p)
1054
self.failIf(osutils.lexists(path),path+" exists")
1056
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1057
"""A helper for callDeprecated and applyDeprecated.
1059
:param a_callable: A callable to call.
1060
:param args: The positional arguments for the callable
1061
:param kwargs: The keyword arguments for the callable
1062
:return: A tuple (warnings, result). result is the result of calling
1063
a_callable(``*args``, ``**kwargs``).
1066
def capture_warnings(msg, cls=None, stacklevel=None):
1067
# we've hooked into a deprecation specific callpath,
1068
# only deprecations should getting sent via it.
1069
self.assertEqual(cls, DeprecationWarning)
1070
local_warnings.append(msg)
1071
original_warning_method = symbol_versioning.warn
1072
symbol_versioning.set_warning_method(capture_warnings)
1074
result = a_callable(*args, **kwargs)
1076
symbol_versioning.set_warning_method(original_warning_method)
1077
return (local_warnings, result)
1079
def applyDeprecated(self, deprecation_format, a_callable, *args, **kwargs):
1080
"""Call a deprecated callable without warning the user.
1082
Note that this only captures warnings raised by symbol_versioning.warn,
1083
not other callers that go direct to the warning module.
1085
To test that a deprecated method raises an error, do something like
1088
self.assertRaises(errors.ReservedId,
1089
self.applyDeprecated,
1090
deprecated_in((1, 5, 0)),
1094
:param deprecation_format: The deprecation format that the callable
1095
should have been deprecated with. This is the same type as the
1096
parameter to deprecated_method/deprecated_function. If the
1097
callable is not deprecated with this format, an assertion error
1099
:param a_callable: A callable to call. This may be a bound method or
1100
a regular function. It will be called with ``*args`` and
1102
:param args: The positional arguments for the callable
1103
:param kwargs: The keyword arguments for the callable
1104
:return: The result of a_callable(``*args``, ``**kwargs``)
1106
call_warnings, result = self._capture_deprecation_warnings(a_callable,
1108
expected_first_warning = symbol_versioning.deprecation_string(
1109
a_callable, deprecation_format)
1110
if len(call_warnings) == 0:
1111
self.fail("No deprecation warning generated by call to %s" %
1113
self.assertEqual(expected_first_warning, call_warnings[0])
1116
def callCatchWarnings(self, fn, *args, **kw):
1117
"""Call a callable that raises python warnings.
1119
The caller's responsible for examining the returned warnings.
1121
If the callable raises an exception, the exception is not
1122
caught and propagates up to the caller. In that case, the list
1123
of warnings is not available.
1125
:returns: ([warning_object, ...], fn_result)
1127
# XXX: This is not perfect, because it completely overrides the
1128
# warnings filters, and some code may depend on suppressing particular
1129
# warnings. It's the easiest way to insulate ourselves from -Werror,
1130
# though. -- Andrew, 20071062
1132
def _catcher(message, category, filename, lineno, file=None):
1133
# despite the name, 'message' is normally(?) a Warning subclass
1135
wlist.append(message)
1136
saved_showwarning = warnings.showwarning
1137
saved_filters = warnings.filters
1139
warnings.showwarning = _catcher
1140
warnings.filters = []
1141
result = fn(*args, **kw)
1143
warnings.showwarning = saved_showwarning
1144
warnings.filters = saved_filters
1145
return wlist, result
1147
def callDeprecated(self, expected, callable, *args, **kwargs):
1148
"""Assert that a callable is deprecated in a particular way.
1150
This is a very precise test for unusual requirements. The
1151
applyDeprecated helper function is probably more suited for most tests
1152
as it allows you to simply specify the deprecation format being used
1153
and will ensure that that is issued for the function being called.
1155
Note that this only captures warnings raised by symbol_versioning.warn,
1156
not other callers that go direct to the warning module. To catch
1157
general warnings, use callCatchWarnings.
1159
:param expected: a list of the deprecation warnings expected, in order
1160
:param callable: The callable to call
1161
:param args: The positional arguments for the callable
1162
:param kwargs: The keyword arguments for the callable
1164
call_warnings, result = self._capture_deprecation_warnings(callable,
1166
self.assertEqual(expected, call_warnings)
1169
def _startLogFile(self):
1170
"""Send bzr and test log messages to a temporary file.
1172
The file is removed as the test is torn down.
60
bzrlib.trace.disable_default_logging()
61
self._enable_file_logging()
64
def _enable_file_logging(self):
1174
65
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
1175
67
self._log_file = os.fdopen(fileno, 'w+')
1176
self._log_memento = bzrlib.trace.push_log_file(self._log_file)
69
hdlr = logging.StreamHandler(self._log_file)
70
hdlr.setLevel(logging.DEBUG)
71
hdlr.setFormatter(logging.Formatter('%(levelname)4.4s %(message)s'))
72
logging.getLogger('').addHandler(hdlr)
73
logging.getLogger('').setLevel(logging.DEBUG)
75
debug('opened log file %s', name)
1177
77
self._log_file_name = name
1178
self.addCleanup(self._finishLogFile)
1180
def _finishLogFile(self):
1181
"""Finished with the log file.
1183
Close the file and delete it, unless setKeepLogfile was called.
1185
if self._log_file is None:
1187
bzrlib.trace.pop_log_file(self._log_memento)
80
logging.getLogger('').removeHandler(self._log_hdlr)
81
bzrlib.trace.enable_default_logging()
82
logging.debug('%s teardown', self.id())
1188
83
self._log_file.close()
1189
self._log_file = None
1190
if not self._keep_log_file:
1191
os.remove(self._log_file_name)
1192
self._log_file_name = None
1194
def setKeepLogfile(self):
1195
"""Make the logfile not be deleted when _finishLogFile is called."""
1196
self._keep_log_file = True
1198
def addCleanup(self, callable):
1199
"""Arrange to run a callable when this case is torn down.
1201
Callables are run in the reverse of the order they are registered,
1202
ie last-in first-out.
1204
if callable in self._cleanups:
1205
raise ValueError("cleanup function %r already registered on %s"
1207
self._cleanups.append(callable)
1209
def _cleanEnvironment(self):
1211
'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
1212
'HOME': os.getcwd(),
1213
'APPDATA': None, # bzr now use Win32 API and don't rely on APPDATA
1214
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
1216
'BZREMAIL': None, # may still be present in the environment
1218
'BZR_PROGRESS_BAR': None,
1221
'SSH_AUTH_SOCK': None,
1225
'https_proxy': None,
1226
'HTTPS_PROXY': None,
1231
# Nobody cares about these ones AFAIK. So far at
1232
# least. If you do (care), please update this comment
1236
'BZR_REMOTE_PATH': None,
1239
self.addCleanup(self._restoreEnvironment)
1240
for name, value in new_env.iteritems():
1241
self._captureVar(name, value)
1243
def _captureVar(self, name, newvalue):
1244
"""Set an environment variable, and reset it when finished."""
1245
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
1247
def _restore_debug_flags(self):
1248
debug.debug_flags.clear()
1249
debug.debug_flags.update(self._preserved_debug_flags)
1251
def _restoreEnvironment(self):
1252
for name, value in self.__old_env.iteritems():
1253
osutils.set_or_unset_env(name, value)
1255
def _restoreHooks(self):
1256
for klass, hooks in self._preserved_hooks.items():
1257
setattr(klass, 'hooks', hooks)
1259
def knownFailure(self, reason):
1260
"""This test has failed for some known reason."""
1261
raise KnownFailure(reason)
1263
def run(self, result=None):
1264
if result is None: result = self.defaultTestResult()
1265
for feature in getattr(self, '_test_needs_features', []):
1266
if not feature.available():
1267
result.startTest(self)
1268
if getattr(result, 'addNotSupported', None):
1269
result.addNotSupported(self, feature)
1271
result.addSuccess(self)
1272
result.stopTest(self)
1275
return unittest.TestCase.run(self, result)
1278
absent_attr = object()
1279
for attr_name in self.attrs_to_keep:
1280
attr = getattr(self, attr_name, absent_attr)
1281
if attr is not absent_attr:
1282
saved_attrs[attr_name] = attr
1283
self.__dict__ = saved_attrs
1287
84
unittest.TestCase.tearDown(self)
1289
def time(self, callable, *args, **kwargs):
1290
"""Run callable and accrue the time it takes to the benchmark time.
1292
If lsprofiling is enabled (i.e. by --lsprof-time to bzr selftest) then
1293
this will cause lsprofile statistics to be gathered and stored in
1296
if self._benchtime is None:
1300
if not self._gather_lsprof_in_benchmarks:
1301
return callable(*args, **kwargs)
1303
# record this benchmark
1304
ret, stats = bzrlib.lsprof.profile(callable, *args, **kwargs)
1306
self._benchcalls.append(((callable, args, kwargs), stats))
1309
self._benchtime += time.time() - start
1311
def _runCleanups(self):
1312
"""Run registered cleanup functions.
1314
This should only be called from TestCase.tearDown.
1316
# TODO: Perhaps this should keep running cleanups even if
1317
# one of them fails?
1319
# Actually pop the cleanups from the list so tearDown running
1320
# twice is safe (this happens for skipped tests).
1321
while self._cleanups:
1322
self._cleanups.pop()()
1324
86
def log(self, *args):
1327
def _get_log(self, keep_log_file=False):
1328
"""Get the log from bzrlib.trace calls from this test.
1330
:param keep_log_file: When True, if the log is still a file on disk
1331
leave it as a file on disk. When False, if the log is still a file
1332
on disk, the log file is deleted and the log preserved as
1334
:return: A string containing the log.
1336
# flush the log file, to get all content
1338
bzrlib.trace._trace_file.flush()
1339
if self._log_contents:
1340
# XXX: this can hardly contain the content flushed above --vila
1342
return self._log_contents
1343
if self._log_file_name is not None:
1344
logfile = open(self._log_file_name)
1346
log_contents = logfile.read()
1349
if not keep_log_file:
1350
self._log_contents = log_contents
1352
os.remove(self._log_file_name)
1354
if sys.platform == 'win32' and e.errno == errno.EACCES:
1355
sys.stderr.write(('Unable to delete log file '
1356
' %r\n' % self._log_file_name))
1361
return "DELETED log file to reduce memory footprint"
1363
def requireFeature(self, feature):
1364
"""This test requires a specific feature is available.
1366
:raises UnavailableFeature: When feature is not available.
1368
if not feature.available():
1369
raise UnavailableFeature(feature)
1371
def _run_bzr_autosplit(self, args, retcode, encoding, stdin,
1373
"""Run bazaar command line, splitting up a string command line."""
1374
if isinstance(args, basestring):
1375
# shlex don't understand unicode strings,
1376
# so args should be plain string (bialix 20070906)
1377
args = list(shlex.split(str(args)))
1378
return self._run_bzr_core(args, retcode=retcode,
1379
encoding=encoding, stdin=stdin, working_dir=working_dir,
1382
def _run_bzr_core(self, args, retcode, encoding, stdin,
1384
if encoding is None:
1385
encoding = bzrlib.user_encoding
1386
stdout = StringIOWrapper()
1387
stderr = StringIOWrapper()
1388
stdout.encoding = encoding
1389
stderr.encoding = encoding
1391
self.log('run bzr: %r', args)
1392
# FIXME: don't call into logging here
90
"""Return as a string the log for this test"""
91
return open(self._log_file_name).read()
94
def capture(self, cmd):
95
"""Shortcut that splits cmd into words, runs, and returns stdout"""
96
return self.run_bzr_captured(cmd.split())[0]
98
def run_bzr_captured(self, argv, retcode=0):
99
"""Invoke bzr and return (result, stdout, stderr).
101
Useful for code that wants to check the contents of the
102
output, the way error messages are presented, etc.
104
This should be the main method for tests that want to exercise the
105
overall behavior of the bzr application (rather than a unit test
106
or a functional test of the library.)
108
Much of the old code runs bzr by forking a new copy of Python, but
109
that is slower, harder to debug, and generally not necessary.
111
This runs bzr through the interface that catches and reports
112
errors, and with logging set to something approximating the
113
default, so that error reporting can be checked.
115
argv -- arguments to invoke bzr
116
retcode -- expected return code, or None for don't-care.
120
self.log('run bzr: %s', ' '.join(argv))
1393
121
handler = logging.StreamHandler(stderr)
122
handler.setFormatter(bzrlib.trace.QuietFormatter())
1394
123
handler.setLevel(logging.INFO)
1395
124
logger = logging.getLogger('')
1396
125
logger.addHandler(handler)
1397
old_ui_factory = ui.ui_factory
1398
ui.ui_factory = TestUIFactory(stdin=stdin, stdout=stdout, stderr=stderr)
1401
if working_dir is not None:
1402
cwd = osutils.getcwd()
1403
os.chdir(working_dir)
1406
result = self.apply_redirected(ui.ui_factory.stdin,
1408
bzrlib.commands.run_bzr_catch_user_errors,
127
result = self.apply_redirected(None, stdout, stderr,
128
bzrlib.commands.run_bzr_catch_errors,
1411
131
logger.removeHandler(handler)
1412
ui.ui_factory = old_ui_factory
1416
132
out = stdout.getvalue()
1417
133
err = stderr.getvalue()
1419
self.log('output:\n%r', out)
135
self.log('output:\n%s', out)
1421
self.log('errors:\n%r', err)
137
self.log('errors:\n%s', err)
1422
138
if retcode is not None:
1423
self.assertEquals(retcode, result,
1424
message='Unexpected return code')
139
self.assertEquals(result, retcode)
1427
def run_bzr(self, args, retcode=0, encoding=None, stdin=None,
1428
working_dir=None, error_regexes=[], output_encoding=None):
142
def run_bzr(self, *args, **kwargs):
1429
143
"""Invoke bzr, as if it were run from the command line.
1431
The argument list should not include the bzr program name - the
1432
first argument is normally the bzr command. Arguments may be
1433
passed in three ways:
1435
1- A list of strings, eg ["commit", "a"]. This is recommended
1436
when the command contains whitespace or metacharacters, or
1437
is built up at run time.
1439
2- A single string, eg "add a". This is the most convenient
1440
for hardcoded commands.
1442
This runs bzr through the interface that catches and reports
1443
errors, and with logging set to something approximating the
1444
default, so that error reporting can be checked.
1446
145
This should be the main method for tests that want to exercise the
1447
146
overall behavior of the bzr application (rather than a unit test
1448
147
or a functional test of the library.)
1450
149
This sends the stdout/stderr results into the test's log,
1451
150
where it may be useful for debugging. See also run_captured.
1453
:keyword stdin: A string to be used as stdin for the command.
1454
:keyword retcode: The status code the command should return;
1456
:keyword working_dir: The directory to run the command in
1457
:keyword error_regexes: A list of expected error messages. If
1458
specified they must be seen in the error output of the command.
1460
out, err = self._run_bzr_autosplit(
1465
working_dir=working_dir,
1467
for regex in error_regexes:
1468
self.assertContainsRe(err, regex)
1471
def run_bzr_error(self, error_regexes, *args, **kwargs):
1472
"""Run bzr, and check that stderr contains the supplied regexes
1474
:param error_regexes: Sequence of regular expressions which
1475
must each be found in the error output. The relative ordering
1477
:param args: command-line arguments for bzr
1478
:param kwargs: Keyword arguments which are interpreted by run_bzr
1479
This function changes the default value of retcode to be 3,
1480
since in most cases this is run when you expect bzr to fail.
1482
:return: (out, err) The actual output of running the command (in case
1483
you want to do more inspection)
1487
# Make sure that commit is failing because there is nothing to do
1488
self.run_bzr_error(['no changes to commit'],
1489
['commit', '-m', 'my commit comment'])
1490
# Make sure --strict is handling an unknown file, rather than
1491
# giving us the 'nothing to do' error
1492
self.build_tree(['unknown'])
1493
self.run_bzr_error(['Commit refused because there are unknown files'],
1494
['commit', --strict', '-m', 'my commit comment'])
1496
kwargs.setdefault('retcode', 3)
1497
kwargs['error_regexes'] = error_regexes
1498
out, err = self.run_bzr(*args, **kwargs)
1501
def run_bzr_subprocess(self, *args, **kwargs):
1502
"""Run bzr in a subprocess for testing.
1504
This starts a new Python interpreter and runs bzr in there.
1505
This should only be used for tests that have a justifiable need for
1506
this isolation: e.g. they are testing startup time, or signal
1507
handling, or early startup code, etc. Subprocess code can't be
1508
profiled or debugged so easily.
1510
:keyword retcode: The status code that is expected. Defaults to 0. If
1511
None is supplied, the status code is not checked.
1512
:keyword env_changes: A dictionary which lists changes to environment
1513
variables. A value of None will unset the env variable.
1514
The values must be strings. The change will only occur in the
1515
child, so you don't need to fix the environment after running.
1516
:keyword universal_newlines: Convert CRLF => LF
1517
:keyword allow_plugins: By default the subprocess is run with
1518
--no-plugins to ensure test reproducibility. Also, it is possible
1519
for system-wide plugins to create unexpected output on stderr,
1520
which can cause unnecessary test failures.
1522
env_changes = kwargs.get('env_changes', {})
1523
working_dir = kwargs.get('working_dir', None)
1524
allow_plugins = kwargs.get('allow_plugins', False)
1526
if isinstance(args[0], list):
1528
elif isinstance(args[0], basestring):
1529
args = list(shlex.split(args[0]))
1531
raise ValueError("passing varargs to run_bzr_subprocess")
1532
process = self.start_bzr_subprocess(args, env_changes=env_changes,
1533
working_dir=working_dir,
1534
allow_plugins=allow_plugins)
1535
# We distinguish between retcode=None and retcode not passed.
1536
supplied_retcode = kwargs.get('retcode', 0)
1537
return self.finish_bzr_subprocess(process, retcode=supplied_retcode,
1538
universal_newlines=kwargs.get('universal_newlines', False),
1541
def start_bzr_subprocess(self, process_args, env_changes=None,
1542
skip_if_plan_to_signal=False,
1544
allow_plugins=False):
1545
"""Start bzr in a subprocess for testing.
1547
This starts a new Python interpreter and runs bzr in there.
1548
This should only be used for tests that have a justifiable need for
1549
this isolation: e.g. they are testing startup time, or signal
1550
handling, or early startup code, etc. Subprocess code can't be
1551
profiled or debugged so easily.
1553
:param process_args: a list of arguments to pass to the bzr executable,
1554
for example ``['--version']``.
1555
:param env_changes: A dictionary which lists changes to environment
1556
variables. A value of None will unset the env variable.
1557
The values must be strings. The change will only occur in the
1558
child, so you don't need to fix the environment after running.
1559
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1561
:param allow_plugins: If False (default) pass --no-plugins to bzr.
1563
:returns: Popen object for the started process.
1565
if skip_if_plan_to_signal:
1566
if not getattr(os, 'kill', None):
1567
raise TestSkipped("os.kill not available.")
1569
if env_changes is None:
1573
def cleanup_environment():
1574
for env_var, value in env_changes.iteritems():
1575
old_env[env_var] = osutils.set_or_unset_env(env_var, value)
1577
def restore_environment():
1578
for env_var, value in old_env.iteritems():
1579
osutils.set_or_unset_env(env_var, value)
1581
bzr_path = self.get_bzr_path()
1584
if working_dir is not None:
1585
cwd = osutils.getcwd()
1586
os.chdir(working_dir)
1589
# win32 subprocess doesn't support preexec_fn
1590
# so we will avoid using it on all platforms, just to
1591
# make sure the code path is used, and we don't break on win32
1592
cleanup_environment()
1593
command = [sys.executable, bzr_path]
1594
if not allow_plugins:
1595
command.append('--no-plugins')
1596
command.extend(process_args)
1597
process = self._popen(command, stdin=PIPE, stdout=PIPE, stderr=PIPE)
1599
restore_environment()
1605
def _popen(self, *args, **kwargs):
1606
"""Place a call to Popen.
1608
Allows tests to override this method to intercept the calls made to
1609
Popen for introspection.
1611
return Popen(*args, **kwargs)
1613
def get_bzr_path(self):
1614
"""Return the path of the 'bzr' executable for this test suite."""
1615
bzr_path = os.path.dirname(os.path.dirname(bzrlib.__file__))+'/bzr'
1616
if not os.path.isfile(bzr_path):
1617
# We are probably installed. Assume sys.argv is the right file
1618
bzr_path = sys.argv[0]
1621
def finish_bzr_subprocess(self, process, retcode=0, send_signal=None,
1622
universal_newlines=False, process_args=None):
1623
"""Finish the execution of process.
1625
:param process: the Popen object returned from start_bzr_subprocess.
1626
:param retcode: The status code that is expected. Defaults to 0. If
1627
None is supplied, the status code is not checked.
1628
:param send_signal: an optional signal to send to the process.
1629
:param universal_newlines: Convert CRLF => LF
1630
:returns: (stdout, stderr)
1632
if send_signal is not None:
1633
os.kill(process.pid, send_signal)
1634
out, err = process.communicate()
1636
if universal_newlines:
1637
out = out.replace('\r\n', '\n')
1638
err = err.replace('\r\n', '\n')
1640
if retcode is not None and retcode != process.returncode:
1641
if process_args is None:
1642
process_args = "(unknown args)"
1643
mutter('Output of bzr %s:\n%s', process_args, out)
1644
mutter('Error for bzr %s:\n%s', process_args, err)
1645
self.fail('Command bzr %s failed with retcode %s != %s'
1646
% (process_args, retcode, process.returncode))
152
retcode = kwargs.pop('retcode', 0)
153
return self.run_bzr_captured(args, retcode)
1649
155
def check_inventory_shape(self, inv, shape):
1650
"""Compare an inventory to a list of expected names.
157
Compare an inventory to a list of expected names.
1652
159
Fail if they are not precisely equal.
1698
205
sys.stderr = real_stderr
1699
206
sys.stdin = real_stdin
1701
def reduceLockdirTimeout(self):
1702
"""Reduce the default lock timeout for the duration of the test, so that
1703
if LockContention occurs during a test, it does so quickly.
1705
Tests that expect to provoke LockContention errors should call this.
1707
orig_timeout = bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS
1709
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = orig_timeout
1710
self.addCleanup(resetTimeout)
1711
bzrlib.lockdir._DEFAULT_TIMEOUT_SECONDS = 0
1713
def make_utf8_encoded_stringio(self, encoding_type=None):
1714
"""Return a StringIOWrapper instance, that will encode Unicode
1717
if encoding_type is None:
1718
encoding_type = 'strict'
1720
output_encoding = 'utf-8'
1721
sio = codecs.getwriter(output_encoding)(sio, errors=encoding_type)
1722
sio.encoding = output_encoding
1726
class TestCaseWithMemoryTransport(TestCase):
1727
"""Common test class for tests that do not need disk resources.
1729
Tests that need disk resources should derive from TestCaseWithTransport.
1731
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
1733
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
1734
a directory which does not exist. This serves to help ensure test isolation
1735
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
1736
must exist. However, TestCaseWithMemoryTransport does not offer local
1737
file defaults for the transport in tests, nor does it obey the command line
1738
override, so tests that accidentally write to the common directory should
1741
:cvar TEST_ROOT: Directory containing all temporary directories, plus
1742
a .bzr directory that stops us ascending higher into the filesystem.
1748
def __init__(self, methodName='runTest'):
1749
# allow test parameterization after test construction and before test
1750
# execution. Variables that the parameterizer sets need to be
1751
# ones that are not set by setUp, or setUp will trash them.
1752
super(TestCaseWithMemoryTransport, self).__init__(methodName)
1753
self.vfs_transport_factory = default_transport
1754
self.transport_server = None
1755
self.transport_readonly_server = None
1756
self.__vfs_server = None
1758
def get_transport(self, relpath=None):
1759
"""Return a writeable transport.
1761
This transport is for the test scratch space relative to
1764
:param relpath: a path relative to the base url.
1766
t = get_transport(self.get_url(relpath))
1767
self.assertFalse(t.is_readonly())
1770
def get_readonly_transport(self, relpath=None):
1771
"""Return a readonly transport for the test scratch space
1773
This can be used to test that operations which should only need
1774
readonly access in fact do not try to write.
1776
:param relpath: a path relative to the base url.
1778
t = get_transport(self.get_readonly_url(relpath))
1779
self.assertTrue(t.is_readonly())
1782
def create_transport_readonly_server(self):
1783
"""Create a transport server from class defined at init.
1785
This is mostly a hook for daughter classes.
1787
return self.transport_readonly_server()
1789
def get_readonly_server(self):
1790
"""Get the server instance for the readonly transport
1792
This is useful for some tests with specific servers to do diagnostics.
1794
if self.__readonly_server is None:
1795
if self.transport_readonly_server is None:
1796
# readonly decorator requested
1797
# bring up the server
1798
self.__readonly_server = ReadonlyServer()
1799
self.__readonly_server.setUp(self.get_vfs_only_server())
1801
self.__readonly_server = self.create_transport_readonly_server()
1802
self.__readonly_server.setUp(self.get_vfs_only_server())
1803
self.addCleanup(self.__readonly_server.tearDown)
1804
return self.__readonly_server
1806
def get_readonly_url(self, relpath=None):
1807
"""Get a URL for the readonly transport.
1809
This will either be backed by '.' or a decorator to the transport
1810
used by self.get_url()
1811
relpath provides for clients to get a path relative to the base url.
1812
These should only be downwards relative, not upwards.
1814
base = self.get_readonly_server().get_url()
1815
return self._adjust_url(base, relpath)
1817
def get_vfs_only_server(self):
1818
"""Get the vfs only read/write server instance.
1820
This is useful for some tests with specific servers that need
1823
For TestCaseWithMemoryTransport this is always a MemoryServer, and there
1824
is no means to override it.
1826
if self.__vfs_server is None:
1827
self.__vfs_server = MemoryServer()
1828
self.__vfs_server.setUp()
1829
self.addCleanup(self.__vfs_server.tearDown)
1830
return self.__vfs_server
1832
def get_server(self):
1833
"""Get the read/write server instance.
1835
This is useful for some tests with specific servers that need
1838
This is built from the self.transport_server factory. If that is None,
1839
then the self.get_vfs_server is returned.
1841
if self.__server is None:
1842
if self.transport_server is None or self.transport_server is self.vfs_transport_factory:
1843
return self.get_vfs_only_server()
1845
# bring up a decorated means of access to the vfs only server.
1846
self.__server = self.transport_server()
1848
self.__server.setUp(self.get_vfs_only_server())
1849
except TypeError, e:
1850
# This should never happen; the try:Except here is to assist
1851
# developers having to update code rather than seeing an
1852
# uninformative TypeError.
1853
raise Exception, "Old server API in use: %s, %s" % (self.__server, e)
1854
self.addCleanup(self.__server.tearDown)
1855
return self.__server
1857
def _adjust_url(self, base, relpath):
1858
"""Get a URL (or maybe a path) for the readwrite transport.
1860
This will either be backed by '.' or to an equivalent non-file based
1862
relpath provides for clients to get a path relative to the base url.
1863
These should only be downwards relative, not upwards.
1865
if relpath is not None and relpath != '.':
1866
if not base.endswith('/'):
1868
# XXX: Really base should be a url; we did after all call
1869
# get_url()! But sometimes it's just a path (from
1870
# LocalAbspathServer), and it'd be wrong to append urlescaped data
1871
# to a non-escaped local path.
1872
if base.startswith('./') or base.startswith('/'):
1875
base += urlutils.escape(relpath)
1878
def get_url(self, relpath=None):
1879
"""Get a URL (or maybe a path) for the readwrite transport.
1881
This will either be backed by '.' or to an equivalent non-file based
1883
relpath provides for clients to get a path relative to the base url.
1884
These should only be downwards relative, not upwards.
1886
base = self.get_server().get_url()
1887
return self._adjust_url(base, relpath)
1889
def get_vfs_only_url(self, relpath=None):
1890
"""Get a URL (or maybe a path for the plain old vfs transport.
1892
This will never be a smart protocol. It always has all the
1893
capabilities of the local filesystem, but it might actually be a
1894
MemoryTransport or some other similar virtual filesystem.
1896
This is the backing transport (if any) of the server returned by
1897
get_url and get_readonly_url.
1899
:param relpath: provides for clients to get a path relative to the base
1900
url. These should only be downwards relative, not upwards.
1903
base = self.get_vfs_only_server().get_url()
1904
return self._adjust_url(base, relpath)
1906
def _create_safety_net(self):
1907
"""Make a fake bzr directory.
1909
This prevents any tests propagating up onto the TEST_ROOT directory's
1912
root = TestCaseWithMemoryTransport.TEST_ROOT
1913
bzrdir.BzrDir.create_standalone_workingtree(root)
1915
def _check_safety_net(self):
1916
"""Check that the safety .bzr directory have not been touched.
1918
_make_test_root have created a .bzr directory to prevent tests from
1919
propagating. This method ensures than a test did not leaked.
1921
root = TestCaseWithMemoryTransport.TEST_ROOT
1922
wt = workingtree.WorkingTree.open(root)
1923
last_rev = wt.last_revision()
1924
if last_rev != 'null:':
1925
# The current test have modified the /bzr directory, we need to
1926
# recreate a new one or all the followng tests will fail.
1927
# If you need to inspect its content uncomment the following line
1928
# import pdb; pdb.set_trace()
1929
_rmtree_temp_dir(root + '/.bzr')
1930
self._create_safety_net()
1931
raise AssertionError('%s/.bzr should not be modified' % root)
1933
def _make_test_root(self):
1934
if TestCaseWithMemoryTransport.TEST_ROOT is None:
1935
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
1936
TestCaseWithMemoryTransport.TEST_ROOT = root
1938
self._create_safety_net()
1940
# The same directory is used by all tests, and we're not
1941
# specifically told when all tests are finished. This will do.
1942
atexit.register(_rmtree_temp_dir, root)
1944
self.addCleanup(self._check_safety_net)
1946
def makeAndChdirToTestDir(self):
1947
"""Create a temporary directories for this one test.
1949
This must set self.test_home_dir and self.test_dir and chdir to
1952
For TestCaseWithMemoryTransport we chdir to the TEST_ROOT for this test.
1954
os.chdir(TestCaseWithMemoryTransport.TEST_ROOT)
1955
self.test_dir = TestCaseWithMemoryTransport.TEST_ROOT
1956
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
1958
def make_branch(self, relpath, format=None):
1959
"""Create a branch on the transport at relpath."""
1960
repo = self.make_repository(relpath, format=format)
1961
return repo.bzrdir.create_branch()
1963
def make_bzrdir(self, relpath, format=None):
1965
# might be a relative or absolute path
1966
maybe_a_url = self.get_url(relpath)
1967
segments = maybe_a_url.rsplit('/', 1)
1968
t = get_transport(maybe_a_url)
1969
if len(segments) > 1 and segments[-1] not in ('', '.'):
1973
if isinstance(format, basestring):
1974
format = bzrdir.format_registry.make_bzrdir(format)
1975
return format.initialize_on_transport(t)
1976
except errors.UninitializableFormat:
1977
raise TestSkipped("Format %s is not initializable." % format)
1979
def make_repository(self, relpath, shared=False, format=None):
1980
"""Create a repository on our default transport at relpath.
1982
Note that relpath must be a relative path, not a full url.
1984
# FIXME: If you create a remoterepository this returns the underlying
1985
# real format, which is incorrect. Actually we should make sure that
1986
# RemoteBzrDir returns a RemoteRepository.
1987
# maybe mbp 20070410
1988
made_control = self.make_bzrdir(relpath, format=format)
1989
return made_control.create_repository(shared=shared)
1991
def make_branch_and_memory_tree(self, relpath, format=None):
1992
"""Create a branch on the default transport and a MemoryTree for it."""
1993
b = self.make_branch(relpath, format=format)
1994
return memorytree.MemoryTree.create_on_branch(b)
1996
def overrideEnvironmentForTesting(self):
1997
os.environ['HOME'] = self.test_home_dir
1998
os.environ['BZR_HOME'] = self.test_home_dir
2001
super(TestCaseWithMemoryTransport, self).setUp()
2002
self._make_test_root()
2003
_currentdir = os.getcwdu()
2004
def _leaveDirectory():
2005
os.chdir(_currentdir)
2006
self.addCleanup(_leaveDirectory)
2007
self.makeAndChdirToTestDir()
2008
self.overrideEnvironmentForTesting()
2009
self.__readonly_server = None
2010
self.__server = None
2011
self.reduceLockdirTimeout()
209
BzrTestBase = TestCase
2014
class TestCaseInTempDir(TestCaseWithMemoryTransport):
212
class TestCaseInTempDir(TestCase):
2015
213
"""Derived class that runs a test within a temporary directory.
2017
215
This is useful for tests that need to create a branch, etc.
2039
232
if contents != expect:
2040
233
self.log("expected: %r" % expect)
2041
234
self.log("actually: %r" % contents)
2042
self.fail("contents of %s not as expected" % filename)
2044
def makeAndChdirToTestDir(self):
2045
"""See TestCaseWithMemoryTransport.makeAndChdirToTestDir().
2047
For TestCaseInTempDir we create a temporary directory based on the test
2048
name and then create two subdirs - test and home under it.
2050
# create a directory within the top level test directory
2051
candidate_dir = osutils.mkdtemp(dir=self.TEST_ROOT)
2052
# now create test and home directories within this dir
2053
self.test_base_dir = candidate_dir
2054
self.test_home_dir = self.test_base_dir + '/home'
2055
os.mkdir(self.test_home_dir)
2056
self.test_dir = self.test_base_dir + '/work'
235
self.fail("contents of %s not as expected")
237
def _make_test_root(self):
238
if TestCaseInTempDir.TEST_ROOT is not None:
242
root = 'test%04d.tmp' % i
246
if e.errno == errno.EEXIST:
251
# successfully created
252
TestCaseInTempDir.TEST_ROOT = os.path.abspath(root)
254
# make a fake bzr directory there to prevent any tests propagating
255
# up onto the source directory's real branch
256
os.mkdir(os.path.join(TestCaseInTempDir.TEST_ROOT, '.bzr'))
259
super(TestCaseInTempDir, self).setUp()
260
self._make_test_root()
261
self._currentdir = os.getcwdu()
262
self.test_dir = os.path.join(self.TEST_ROOT, self.id())
2057
263
os.mkdir(self.test_dir)
2058
264
os.chdir(self.test_dir)
2059
# put name of test inside
2060
f = file(self.test_base_dir + '/name', 'w')
2065
self.addCleanup(self.deleteTestDir)
2067
def deleteTestDir(self):
2068
os.chdir(self.TEST_ROOT)
2069
_rmtree_temp_dir(self.test_base_dir)
2071
def build_tree(self, shape, line_endings='binary', transport=None):
267
os.chdir(self._currentdir)
268
super(TestCaseInTempDir, self).tearDown()
270
def build_tree(self, shape):
2072
271
"""Build a test tree according to a pattern.
2074
273
shape is a sequence of file specifications. If the final
2075
274
character is '/', a directory is created.
2077
This assumes that all the elements in the tree being built are new.
2079
276
This doesn't add anything to a branch.
2081
:type shape: list or tuple.
2082
:param line_endings: Either 'binary' or 'native'
2083
in binary mode, exact contents are written in native mode, the
2084
line endings match the default platform endings.
2085
:param transport: A transport to write to, for building trees on VFS's.
2086
If the transport is readonly or None, "." is opened automatically.
2089
if type(shape) not in (list, tuple):
2090
raise AssertionError("Parameter 'shape' should be "
2091
"a list or a tuple. Got %r instead" % (shape,))
2092
# It's OK to just create them using forward slashes on windows.
2093
if transport is None or transport.is_readonly():
2094
transport = get_transport(".")
278
# XXX: It's OK to just create them using forward slashes on windows?
2095
279
for name in shape:
2096
self.assert_(isinstance(name, basestring))
280
assert isinstance(name, basestring)
2097
281
if name[-1] == '/':
2098
transport.mkdir(urlutils.escape(name[:-1]))
2100
if line_endings == 'binary':
2102
elif line_endings == 'native':
2105
raise errors.BzrError(
2106
'Invalid line ending request %r' % line_endings)
2107
content = "contents of %s%s" % (name.encode('utf-8'), end)
2108
transport.put_bytes_non_atomic(urlutils.escape(name), content)
2110
def build_tree_contents(self, shape):
2111
build_tree_contents(shape)
2113
def assertInWorkingTree(self, path, root_path='.', tree=None):
2114
"""Assert whether path or paths are in the WorkingTree"""
2116
tree = workingtree.WorkingTree.open(root_path)
2117
if not isinstance(path, basestring):
2119
self.assertInWorkingTree(p,tree=tree)
2121
self.assertIsNot(tree.path2id(path), None,
2122
path+' not in working tree.')
2124
def assertNotInWorkingTree(self, path, root_path='.', tree=None):
2125
"""Assert whether path or paths are not in the WorkingTree"""
2127
tree = workingtree.WorkingTree.open(root_path)
2128
if not isinstance(path, basestring):
2130
self.assertNotInWorkingTree(p,tree=tree)
2132
self.assertIs(tree.path2id(path), None, path+' in working tree.')
2135
class TestCaseWithTransport(TestCaseInTempDir):
2136
"""A test case that provides get_url and get_readonly_url facilities.
2138
These back onto two transport servers, one for readonly access and one for
2141
If no explicit class is provided for readonly access, a
2142
ReadonlyTransportDecorator is used instead which allows the use of non disk
2143
based read write transports.
2145
If an explicit class is provided for readonly access, that server and the
2146
readwrite one must both define get_url() as resolving to os.getcwd().
2149
def get_vfs_only_server(self):
2150
"""See TestCaseWithMemoryTransport.
2152
This is useful for some tests with specific servers that need
2155
if self.__vfs_server is None:
2156
self.__vfs_server = self.vfs_transport_factory()
2157
self.__vfs_server.setUp()
2158
self.addCleanup(self.__vfs_server.tearDown)
2159
return self.__vfs_server
2161
def make_branch_and_tree(self, relpath, format=None):
2162
"""Create a branch on the transport and a tree locally.
2164
If the transport is not a LocalTransport, the Tree can't be created on
2165
the transport. In that case if the vfs_transport_factory is
2166
LocalURLServer the working tree is created in the local
2167
directory backing the transport, and the returned tree's branch and
2168
repository will also be accessed locally. Otherwise a lightweight
2169
checkout is created and returned.
2171
:param format: The BzrDirFormat.
2172
:returns: the WorkingTree.
2174
# TODO: always use the local disk path for the working tree,
2175
# this obviously requires a format that supports branch references
2176
# so check for that by checking bzrdir.BzrDirFormat.get_default_format()
2178
b = self.make_branch(relpath, format=format)
2180
return b.bzrdir.create_workingtree()
2181
except errors.NotLocalUrl:
2182
# We can only make working trees locally at the moment. If the
2183
# transport can't support them, then we keep the non-disk-backed
2184
# branch and create a local checkout.
2185
if self.vfs_transport_factory is LocalURLServer:
2186
# the branch is colocated on disk, we cannot create a checkout.
2187
# hopefully callers will expect this.
2188
local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
2189
return local_controldir.create_workingtree()
2191
return b.create_checkout(relpath, lightweight=True)
2193
def assertIsDirectory(self, relpath, transport):
2194
"""Assert that relpath within transport is a directory.
2196
This may not be possible on all transports; in that case it propagates
2197
a TransportNotPossible.
2200
mode = transport.stat(relpath).st_mode
2201
except errors.NoSuchFile:
2202
self.fail("path %s is not a directory; no such file"
2204
if not stat.S_ISDIR(mode):
2205
self.fail("path %s is not a directory; has mode %#o"
2208
def assertTreesEqual(self, left, right):
2209
"""Check that left and right have the same content and properties."""
2210
# we use a tree delta to check for equality of the content, and we
2211
# manually check for equality of other things such as the parents list.
2212
self.assertEqual(left.get_parent_ids(), right.get_parent_ids())
2213
differences = left.changes_from(right)
2214
self.assertFalse(differences.has_changed(),
2215
"Trees %r and %r are different: %r" % (left, right, differences))
2218
super(TestCaseWithTransport, self).setUp()
2219
self.__vfs_server = None
2222
class ChrootedTestCase(TestCaseWithTransport):
2223
"""A support class that provides readonly urls outside the local namespace.
2225
This is done by checking if self.transport_server is a MemoryServer. if it
2226
is then we are chrooted already, if it is not then an HttpServer is used
2229
TODO RBC 20060127: make this an option to TestCaseWithTransport so it can
2230
be used without needed to redo it when a different
2231
subclass is in use ?
2235
super(ChrootedTestCase, self).setUp()
2236
if not self.vfs_transport_factory == MemoryServer:
2237
self.transport_readonly_server = HttpServer
2240
def condition_id_re(pattern):
2241
"""Create a condition filter which performs a re check on a test's id.
2243
:param pattern: A regular expression string.
2244
:return: A callable that returns True if the re matches.
2246
filter_re = re.compile(pattern)
2247
def condition(test):
2249
return filter_re.search(test_id)
2253
def condition_isinstance(klass_or_klass_list):
2254
"""Create a condition filter which returns isinstance(param, klass).
2256
:return: A callable which when called with one parameter obj return the
2257
result of isinstance(obj, klass_or_klass_list).
2260
return isinstance(obj, klass_or_klass_list)
2264
def condition_id_in_list(id_list):
2265
"""Create a condition filter which verify that test's id in a list.
2267
:param id_list: A TestIdList object.
2268
:return: A callable that returns True if the test's id appears in the list.
2270
def condition(test):
2271
return id_list.includes(test.id())
2275
def condition_id_startswith(start):
2276
"""Create a condition filter verifying that test's id starts with a string.
2278
:param start: A string.
2279
:return: A callable that returns True if the test's id starts with the
2282
def condition(test):
2283
return test.id().startswith(start)
2287
def exclude_tests_by_condition(suite, condition):
2288
"""Create a test suite which excludes some tests from suite.
2290
:param suite: The suite to get tests from.
2291
:param condition: A callable whose result evaluates True when called with a
2292
test case which should be excluded from the result.
2293
:return: A suite which contains the tests found in suite that fail
2297
for test in iter_suite_tests(suite):
2298
if not condition(test):
2300
return TestUtil.TestSuite(result)
2303
def filter_suite_by_condition(suite, condition):
2304
"""Create a test suite by filtering another one.
2306
:param suite: The source suite.
2307
:param condition: A callable whose result evaluates True when called with a
2308
test case which should be included in the result.
2309
:return: A suite which contains the tests found in suite that pass
2313
for test in iter_suite_tests(suite):
2316
return TestUtil.TestSuite(result)
2319
def filter_suite_by_re(suite, pattern):
2320
"""Create a test suite by filtering another one.
2322
:param suite: the source suite
2323
:param pattern: pattern that names must match
2324
:returns: the newly created suite
2326
condition = condition_id_re(pattern)
2327
result_suite = filter_suite_by_condition(suite, condition)
2331
def filter_suite_by_id_list(suite, test_id_list):
2332
"""Create a test suite by filtering another one.
2334
:param suite: The source suite.
2335
:param test_id_list: A list of the test ids to keep as strings.
2336
:returns: the newly created suite
2338
condition = condition_id_in_list(test_id_list)
2339
result_suite = filter_suite_by_condition(suite, condition)
2343
def filter_suite_by_id_startswith(suite, start):
2344
"""Create a test suite by filtering another one.
2346
:param suite: The source suite.
2347
:param start: A string the test id must start with.
2348
:returns: the newly created suite
2350
condition = condition_id_startswith(start)
2351
result_suite = filter_suite_by_condition(suite, condition)
2355
def exclude_tests_by_re(suite, pattern):
2356
"""Create a test suite which excludes some tests from suite.
2358
:param suite: The suite to get tests from.
2359
:param pattern: A regular expression string. Test ids that match this
2360
pattern will be excluded from the result.
2361
:return: A TestSuite that contains all the tests from suite without the
2362
tests that matched pattern. The order of tests is the same as it was in
2365
return exclude_tests_by_condition(suite, condition_id_re(pattern))
2368
def preserve_input(something):
2369
"""A helper for performing test suite transformation chains.
2371
:param something: Anything you want to preserve.
2377
def randomize_suite(suite):
2378
"""Return a new TestSuite with suite's tests in random order.
2380
The tests in the input suite are flattened into a single suite in order to
2381
accomplish this. Any nested TestSuites are removed to provide global
2384
tests = list(iter_suite_tests(suite))
2385
random.shuffle(tests)
2386
return TestUtil.TestSuite(tests)
2389
def split_suite_by_condition(suite, condition):
2390
"""Split a test suite into two by a condition.
2392
:param suite: The suite to split.
2393
:param condition: The condition to match on. Tests that match this
2394
condition are returned in the first test suite, ones that do not match
2395
are in the second suite.
2396
:return: A tuple of two test suites, where the first contains tests from
2397
suite matching the condition, and the second contains the remainder
2398
from suite. The order within each output suite is the same as it was in
2403
for test in iter_suite_tests(suite):
2405
matched.append(test)
2407
did_not_match.append(test)
2408
return TestUtil.TestSuite(matched), TestUtil.TestSuite(did_not_match)
2411
def split_suite_by_re(suite, pattern):
2412
"""Split a test suite into two by a regular expression.
2414
:param suite: The suite to split.
2415
:param pattern: A regular expression string. Test ids that match this
2416
pattern will be in the first test suite returned, and the others in the
2417
second test suite returned.
2418
:return: A tuple of two test suites, where the first contains tests from
2419
suite matching pattern, and the second contains the remainder from
2420
suite. The order within each output suite is the same as it was in
2423
return split_suite_by_condition(suite, condition_id_re(pattern))
2426
def run_suite(suite, name='test', verbose=False, pattern=".*",
2427
stop_on_failure=False,
2428
transport=None, lsprof_timed=None, bench_history=None,
2429
matching_tests_first=None,
2432
exclude_pattern=None,
2434
TestCase._gather_lsprof_in_benchmarks = lsprof_timed
2439
runner = TextTestRunner(stream=sys.stdout,
2441
verbosity=verbosity,
2442
bench_history=bench_history,
2443
list_only=list_only,
2445
runner.stop_on_failure=stop_on_failure
2446
# Initialise the random number generator and display the seed used.
2447
# We convert the seed to a long to make it reuseable across invocations.
2448
random_order = False
2449
if random_seed is not None:
2451
if random_seed == "now":
2452
random_seed = long(time.time())
2454
# Convert the seed to a long if we can
2456
random_seed = long(random_seed)
2459
runner.stream.writeln("Randomizing test order using seed %s\n" %
2461
random.seed(random_seed)
2462
# Customise the list of tests if requested
2463
if exclude_pattern is not None:
2464
suite = exclude_tests_by_re(suite, exclude_pattern)
2466
order_changer = randomize_suite
2468
order_changer = preserve_input
2469
if pattern != '.*' or random_order:
2470
if matching_tests_first:
2471
suites = map(order_changer, split_suite_by_re(suite, pattern))
2472
suite = TestUtil.TestSuite(suites)
2474
suite = order_changer(filter_suite_by_re(suite, pattern))
2476
result = runner.run(suite)
2479
return result.wasStrictlySuccessful()
2481
return result.wasSuccessful()
2484
# Controlled by "bzr selftest -E=..." option
2485
selftest_debug_flags = set()
2488
def selftest(verbose=False, pattern=".*", stop_on_failure=True,
2490
test_suite_factory=None,
2493
matching_tests_first=None,
2496
exclude_pattern=None,
2502
"""Run the whole test suite under the enhanced runner"""
2503
# XXX: Very ugly way to do this...
2504
# Disable warning about old formats because we don't want it to disturb
2505
# any blackbox tests.
2506
from bzrlib import repository
2507
repository._deprecation_warning_done = True
2509
global default_transport
2510
if transport is None:
2511
transport = default_transport
2512
old_transport = default_transport
2513
default_transport = transport
2514
global selftest_debug_flags
2515
old_debug_flags = selftest_debug_flags
2516
if debug_flags is not None:
2517
selftest_debug_flags = set(debug_flags)
2519
if load_list is None:
2522
keep_only = load_test_id_list(load_list)
2523
if test_suite_factory is None:
2524
suite = test_suite(keep_only, starting_with)
2526
suite = test_suite_factory()
2527
return run_suite(suite, 'testbzr', verbose=verbose, pattern=pattern,
2528
stop_on_failure=stop_on_failure,
2529
transport=transport,
2530
lsprof_timed=lsprof_timed,
2531
bench_history=bench_history,
2532
matching_tests_first=matching_tests_first,
2533
list_only=list_only,
2534
random_seed=random_seed,
2535
exclude_pattern=exclude_pattern,
2538
default_transport = old_transport
2539
selftest_debug_flags = old_debug_flags
2542
def load_test_id_list(file_name):
2543
"""Load a test id list from a text file.
2545
The format is one test id by line. No special care is taken to impose
2546
strict rules, these test ids are used to filter the test suite so a test id
2547
that do not match an existing test will do no harm. This allows user to add
2548
comments, leave blank lines, etc.
2552
ftest = open(file_name, 'rt')
2554
if e.errno != errno.ENOENT:
2557
raise errors.NoSuchFile(file_name)
2559
for test_name in ftest.readlines():
2560
test_list.append(test_name.strip())
2565
def suite_matches_id_list(test_suite, id_list):
2566
"""Warns about tests not appearing or appearing more than once.
2568
:param test_suite: A TestSuite object.
2569
:param test_id_list: The list of test ids that should be found in
2572
:return: (absents, duplicates) absents is a list containing the test found
2573
in id_list but not in test_suite, duplicates is a list containing the
2574
test found multiple times in test_suite.
2576
When using a prefined test id list, it may occurs that some tests do not
2577
exist anymore or that some tests use the same id. This function warns the
2578
tester about potential problems in his workflow (test lists are volatile)
2579
or in the test suite itself (using the same id for several tests does not
2580
help to localize defects).
2582
# Build a dict counting id occurrences
2584
for test in iter_suite_tests(test_suite):
2586
tests[id] = tests.get(id, 0) + 1
2591
occurs = tests.get(id, 0)
2593
not_found.append(id)
2595
duplicates.append(id)
2597
return not_found, duplicates
2600
class TestIdList(object):
2601
"""Test id list to filter a test suite.
2603
Relying on the assumption that test ids are built as:
2604
<module>[.<class>.<method>][(<param>+)], <module> being in python dotted
2605
notation, this class offers methods to :
2606
- avoid building a test suite for modules not refered to in the test list,
2607
- keep only the tests listed from the module test suite.
2610
def __init__(self, test_id_list):
2611
# When a test suite needs to be filtered against us we compare test ids
2612
# for equality, so a simple dict offers a quick and simple solution.
2613
self.tests = dict().fromkeys(test_id_list, True)
2615
# While unittest.TestCase have ids like:
2616
# <module>.<class>.<method>[(<param+)],
2617
# doctest.DocTestCase can have ids like:
2620
# <module>.<function>
2621
# <module>.<class>.<method>
2623
# Since we can't predict a test class from its name only, we settle on
2624
# a simple constraint: a test id always begins with its module name.
2627
for test_id in test_id_list:
2628
parts = test_id.split('.')
2629
mod_name = parts.pop(0)
2630
modules[mod_name] = True
2632
mod_name += '.' + part
2633
modules[mod_name] = True
2634
self.modules = modules
2636
def refers_to(self, module_name):
2637
"""Is there tests for the module or one of its sub modules."""
2638
return self.modules.has_key(module_name)
2640
def includes(self, test_id):
2641
return self.tests.has_key(test_id)
2644
def test_suite(keep_only=None, starting_with=None):
2645
"""Build and return TestSuite for the whole of bzrlib.
2647
:param keep_only: A list of test ids limiting the suite returned.
2649
:param starting_with: An id limiting the suite returned to the tests
2652
This function can be replaced if you need to change the default test
2653
suite on a global basis, but it is not encouraged.
2657
'bzrlib.util.tests.test_bencode',
2658
'bzrlib.tests.blackbox',
2659
'bzrlib.tests.branch_implementations',
2660
'bzrlib.tests.bzrdir_implementations',
2661
'bzrlib.tests.commands',
2662
'bzrlib.tests.inventory_implementations',
2663
'bzrlib.tests.interrepository_implementations',
2664
'bzrlib.tests.intertree_implementations',
2665
'bzrlib.tests.interversionedfile_implementations',
2666
'bzrlib.tests.per_lock',
2667
'bzrlib.tests.repository_implementations',
2668
'bzrlib.tests.revisionstore_implementations',
2669
'bzrlib.tests.test__dirstate_helpers',
2670
'bzrlib.tests.test_ancestry',
2671
'bzrlib.tests.test_annotate',
2672
'bzrlib.tests.test_api',
2673
'bzrlib.tests.test_atomicfile',
2674
'bzrlib.tests.test_bad_files',
2675
'bzrlib.tests.test_bisect_multi',
2676
'bzrlib.tests.test_branch',
2677
'bzrlib.tests.test_branchbuilder',
2678
'bzrlib.tests.test_bugtracker',
2679
'bzrlib.tests.test_bundle',
2680
'bzrlib.tests.test_bzrdir',
2681
'bzrlib.tests.test_cache_utf8',
2682
'bzrlib.tests.test_commands',
2683
'bzrlib.tests.test_commit',
2684
'bzrlib.tests.test_commit_merge',
2685
'bzrlib.tests.test_config',
2686
'bzrlib.tests.test_conflicts',
2687
'bzrlib.tests.test_counted_lock',
2688
'bzrlib.tests.test_decorators',
2689
'bzrlib.tests.test_delta',
2690
'bzrlib.tests.test_deprecated_graph',
2691
'bzrlib.tests.test_diff',
2692
'bzrlib.tests.test_dirstate',
2693
'bzrlib.tests.test_directory_service',
2694
'bzrlib.tests.test_email_message',
2695
'bzrlib.tests.test_errors',
2696
'bzrlib.tests.test_escaped_store',
2697
'bzrlib.tests.test_extract',
2698
'bzrlib.tests.test_fetch',
2699
'bzrlib.tests.test_ftp_transport',
2700
'bzrlib.tests.test_generate_docs',
2701
'bzrlib.tests.test_generate_ids',
2702
'bzrlib.tests.test_globbing',
2703
'bzrlib.tests.test_gpg',
2704
'bzrlib.tests.test_graph',
2705
'bzrlib.tests.test_hashcache',
2706
'bzrlib.tests.test_help',
2707
'bzrlib.tests.test_hooks',
2708
'bzrlib.tests.test_http',
2709
'bzrlib.tests.test_http_implementations',
2710
'bzrlib.tests.test_http_response',
2711
'bzrlib.tests.test_https_ca_bundle',
2712
'bzrlib.tests.test_identitymap',
2713
'bzrlib.tests.test_ignores',
2714
'bzrlib.tests.test_index',
2715
'bzrlib.tests.test_info',
2716
'bzrlib.tests.test_inv',
2717
'bzrlib.tests.test_knit',
2718
'bzrlib.tests.test_lazy_import',
2719
'bzrlib.tests.test_lazy_regex',
2720
'bzrlib.tests.test_lockdir',
2721
'bzrlib.tests.test_lockable_files',
2722
'bzrlib.tests.test_log',
2723
'bzrlib.tests.test_lsprof',
2724
'bzrlib.tests.test_lru_cache',
2725
'bzrlib.tests.test_mail_client',
2726
'bzrlib.tests.test_memorytree',
2727
'bzrlib.tests.test_merge',
2728
'bzrlib.tests.test_merge3',
2729
'bzrlib.tests.test_merge_core',
2730
'bzrlib.tests.test_merge_directive',
2731
'bzrlib.tests.test_missing',
2732
'bzrlib.tests.test_msgeditor',
2733
'bzrlib.tests.test_multiparent',
2734
'bzrlib.tests.test_mutabletree',
2735
'bzrlib.tests.test_nonascii',
2736
'bzrlib.tests.test_options',
2737
'bzrlib.tests.test_osutils',
2738
'bzrlib.tests.test_osutils_encodings',
2739
'bzrlib.tests.test_pack',
2740
'bzrlib.tests.test_patch',
2741
'bzrlib.tests.test_patches',
2742
'bzrlib.tests.test_permissions',
2743
'bzrlib.tests.test_plugins',
2744
'bzrlib.tests.test_progress',
2745
'bzrlib.tests.test_read_bundle',
2746
'bzrlib.tests.test_reconfigure',
2747
'bzrlib.tests.test_reconcile',
2748
'bzrlib.tests.test_registry',
2749
'bzrlib.tests.test_remote',
2750
'bzrlib.tests.test_repository',
2751
'bzrlib.tests.per_repository_reference',
2752
'bzrlib.tests.test_revert',
2753
'bzrlib.tests.test_revision',
2754
'bzrlib.tests.test_revisionspec',
2755
'bzrlib.tests.test_revisiontree',
2756
'bzrlib.tests.test_rio',
2757
'bzrlib.tests.test_sampler',
2758
'bzrlib.tests.test_selftest',
2759
'bzrlib.tests.test_setup',
2760
'bzrlib.tests.test_sftp_transport',
2761
'bzrlib.tests.test_smart',
2762
'bzrlib.tests.test_smart_add',
2763
'bzrlib.tests.test_smart_transport',
2764
'bzrlib.tests.test_smtp_connection',
2765
'bzrlib.tests.test_source',
2766
'bzrlib.tests.test_ssh_transport',
2767
'bzrlib.tests.test_status',
2768
'bzrlib.tests.test_store',
2769
'bzrlib.tests.test_strace',
2770
'bzrlib.tests.test_subsume',
2771
'bzrlib.tests.test_switch',
2772
'bzrlib.tests.test_symbol_versioning',
2773
'bzrlib.tests.test_tag',
2774
'bzrlib.tests.test_testament',
2775
'bzrlib.tests.test_textfile',
2776
'bzrlib.tests.test_textmerge',
2777
'bzrlib.tests.test_timestamp',
2778
'bzrlib.tests.test_trace',
2779
'bzrlib.tests.test_transactions',
2780
'bzrlib.tests.test_transform',
2781
'bzrlib.tests.test_transport',
2782
'bzrlib.tests.test_transport_implementations',
2783
'bzrlib.tests.test_tree',
2784
'bzrlib.tests.test_treebuilder',
2785
'bzrlib.tests.test_tsort',
2786
'bzrlib.tests.test_tuned_gzip',
2787
'bzrlib.tests.test_ui',
2788
'bzrlib.tests.test_uncommit',
2789
'bzrlib.tests.test_upgrade',
2790
'bzrlib.tests.test_urlutils',
2791
'bzrlib.tests.test_versionedfile',
2792
'bzrlib.tests.test_version',
2793
'bzrlib.tests.test_version_info',
2794
'bzrlib.tests.test_weave',
2795
'bzrlib.tests.test_whitebox',
2796
'bzrlib.tests.test_win32utils',
2797
'bzrlib.tests.test_workingtree',
2798
'bzrlib.tests.test_workingtree_4',
2799
'bzrlib.tests.test_wsgi',
2800
'bzrlib.tests.test_xml',
2801
'bzrlib.tests.tree_implementations',
2802
'bzrlib.tests.workingtree_implementations',
285
print >>f, "contents of", name
289
class MetaTestLog(TestCase):
290
def test_logging(self):
291
"""Test logs are captured when a test fails."""
292
logging.info('an info message')
293
warning('something looks dodgy...')
294
logging.debug('hello, test is running')
298
def selftest(verbose=False, pattern=".*"):
299
return testsweet.run_suite(test_suite(), 'testbzr', verbose=verbose, pattern=pattern)
303
from bzrlib.selftest.TestUtil import TestLoader, TestSuite
304
import bzrlib, bzrlib.store, bzrlib.inventory, bzrlib.branch
305
import bzrlib.osutils, bzrlib.commands, bzrlib.merge3, bzrlib.plugin
306
from doctest import DocTestSuite
308
global MODULES_TO_TEST, MODULES_TO_DOCTEST
311
['bzrlib.selftest.MetaTestLog',
312
'bzrlib.selftest.test_parent',
313
'bzrlib.selftest.testinv',
314
'bzrlib.selftest.testfetch',
315
'bzrlib.selftest.versioning',
316
'bzrlib.selftest.whitebox',
317
'bzrlib.selftest.testmerge3',
318
'bzrlib.selftest.testmerge',
319
'bzrlib.selftest.testhashcache',
320
'bzrlib.selftest.teststatus',
321
'bzrlib.selftest.testlog',
322
'bzrlib.selftest.blackbox',
323
'bzrlib.selftest.testrevisionnamespaces',
324
'bzrlib.selftest.testbranch',
325
'bzrlib.selftest.testremotebranch',
326
'bzrlib.selftest.testrevision',
327
'bzrlib.selftest.test_revision_info',
328
'bzrlib.selftest.test_merge_core',
329
'bzrlib.selftest.test_smart_add',
330
'bzrlib.selftest.testdiff',
331
'bzrlib.selftest.test_xml',
333
'bzrlib.selftest.teststore',
334
'bzrlib.selftest.testgraph',
2805
loader = TestUtil.TestLoader()
2807
if starting_with is not None:
2808
# We take precedence over keep_only because *at loading time* using
2809
# both options means we will load less tests for the same final result.
2810
def interesting_module(name):
2812
# Either the module name starts with the specified string
2813
name.startswith(starting_with)
2814
# or it may contain tests starting with the specified string
2815
or starting_with.startswith(name)
2817
loader = TestUtil.FilteredByModuleTestLoader(interesting_module)
2819
elif keep_only is not None:
2820
id_filter = TestIdList(keep_only)
2821
loader = TestUtil.FilteredByModuleTestLoader(id_filter.refers_to)
2822
def interesting_module(name):
2823
return id_filter.refers_to(name)
2826
loader = TestUtil.TestLoader()
2827
def interesting_module(name):
2828
# No filtering, all modules are interesting
2831
suite = loader.suiteClass()
2833
# modules building their suite with loadTestsFromModuleNames
2834
suite.addTest(loader.loadTestsFromModuleNames(testmod_names))
2836
modules_to_doctest = [
2841
'bzrlib.iterablefile',
2846
'bzrlib.symbol_versioning',
2849
'bzrlib.version_info_formats.format_custom',
2852
for mod in modules_to_doctest:
2853
if not interesting_module(mod):
2854
# No tests to keep here, move along
2857
doc_suite = doctest.DocTestSuite(mod)
2858
except ValueError, e:
2859
print '**failed to get doctest for: %s\n%s' % (mod, e)
2861
suite.addTest(doc_suite)
2863
default_encoding = sys.getdefaultencoding()
2864
for name, plugin in bzrlib.plugin.plugins().items():
2865
if not interesting_module(plugin.module.__name__):
2867
plugin_suite = plugin.test_suite()
2868
# We used to catch ImportError here and turn it into just a warning,
2869
# but really if you don't have --no-plugins this should be a failure.
2870
# mbp 20080213 - see http://bugs.launchpad.net/bugs/189771
2871
if plugin_suite is None:
2872
plugin_suite = plugin.load_plugin_tests(loader)
2873
if plugin_suite is not None:
2874
suite.addTest(plugin_suite)
2875
if default_encoding != sys.getdefaultencoding():
2876
bzrlib.trace.warning(
2877
'Plugin "%s" tried to reset default encoding to: %s', name,
2878
sys.getdefaultencoding())
2880
sys.setdefaultencoding(default_encoding)
2882
if starting_with is not None:
2883
suite = filter_suite_by_id_startswith(suite, starting_with)
2885
if keep_only is not None:
2886
# Now that the referred modules have loaded their tests, keep only the
2888
suite = filter_suite_by_id_list(suite, id_filter)
2889
# Do some sanity checks on the id_list filtering
2890
not_found, duplicates = suite_matches_id_list(suite, keep_only)
2891
if starting_with is not None:
2892
# The tester has used both keep_only and starting_with, so he is
2893
# already aware that some tests are excluded from the list, there
2894
# is no need to tell him which.
2897
# Some tests mentioned in the list are not in the test suite. The
2898
# list may be out of date, report to the tester.
2899
for id in not_found:
2900
bzrlib.trace.warning('"%s" not found in the test suite', id)
2901
for id in duplicates:
2902
bzrlib.trace.warning('"%s" is used as an id by several tests', id)
2907
def multiply_tests_from_modules(module_name_list, scenario_iter, loader=None):
2908
"""Adapt all tests in some given modules to given scenarios.
2910
This is the recommended public interface for test parameterization.
2911
Typically the test_suite() method for a per-implementation test
2912
suite will call multiply_tests_from_modules and return the
2915
:param module_name_list: List of fully-qualified names of test
2917
:param scenario_iter: Iterable of pairs of (scenario_name,
2918
scenario_param_dict).
2919
:param loader: If provided, will be used instead of a new
2920
bzrlib.tests.TestLoader() instance.
2922
This returns a new TestSuite containing the cross product of
2923
all the tests in all the modules, each repeated for each scenario.
2924
Each test is adapted by adding the scenario name at the end
2925
of its name, and updating the test object's __dict__ with the
2926
scenario_param_dict.
2928
>>> r = multiply_tests_from_modules(
2929
... ['bzrlib.tests.test_sampler'],
2930
... [('one', dict(param=1)),
2931
... ('two', dict(param=2))])
2932
>>> tests = list(iter_suite_tests(r))
2936
'bzrlib.tests.test_sampler.DemoTest.test_nothing(one)'
2942
# XXX: Isn't load_tests() a better way to provide the same functionality
2943
# without forcing a predefined TestScenarioApplier ? --vila 080215
2945
loader = TestUtil.TestLoader()
2947
suite = loader.suiteClass()
2949
adapter = TestScenarioApplier()
2950
adapter.scenarios = list(scenario_iter)
2951
adapt_modules(module_name_list, adapter, loader, suite)
2955
def multiply_scenarios(scenarios_left, scenarios_right):
2956
"""Multiply two sets of scenarios.
2958
:returns: the cartesian product of the two sets of scenarios, that is
2959
a scenario for every possible combination of a left scenario and a
2963
('%s,%s' % (left_name, right_name),
2964
dict(left_dict.items() + right_dict.items()))
2965
for left_name, left_dict in scenarios_left
2966
for right_name, right_dict in scenarios_right]
2970
def adapt_modules(mods_list, adapter, loader, suite):
2971
"""Adapt the modules in mods_list using adapter and add to suite."""
2972
tests = loader.loadTestsFromModuleNames(mods_list)
2973
adapt_tests(tests, adapter, suite)
2976
def adapt_tests(tests_list, adapter, suite):
2977
"""Adapt the tests in tests_list using adapter and add to suite."""
2978
for test in iter_suite_tests(tests_list):
2979
suite.addTests(adapter.adapt(test))
2982
def _rmtree_temp_dir(dirname):
2983
# If LANG=C we probably have created some bogus paths
2984
# which rmtree(unicode) will fail to delete
2985
# so make sure we are using rmtree(str) to delete everything
2986
# except on win32, where rmtree(str) will fail
2987
# since it doesn't have the property of byte-stream paths
2988
# (they are either ascii or mbcs)
2989
if sys.platform == 'win32':
2990
# make sure we are using the unicode win32 api
2991
dirname = unicode(dirname)
2993
dirname = dirname.encode(sys.getfilesystemencoding())
2995
osutils.rmtree(dirname)
2997
if sys.platform == 'win32' and e.errno == errno.EACCES:
2998
sys.stderr.write(('Permission denied: '
2999
'unable to remove testing dir '
3000
'%s\n' % os.path.basename(dirname)))
3005
class Feature(object):
3006
"""An operating system Feature."""
3009
self._available = None
3011
def available(self):
3012
"""Is the feature available?
3014
:return: True if the feature is available.
3016
if self._available is None:
3017
self._available = self._probe()
3018
return self._available
3021
"""Implement this method in concrete features.
3023
:return: True if the feature is available.
3025
raise NotImplementedError
3028
if getattr(self, 'feature_name', None):
3029
return self.feature_name()
3030
return self.__class__.__name__
3033
class _SymlinkFeature(Feature):
3036
return osutils.has_symlinks()
3038
def feature_name(self):
3041
SymlinkFeature = _SymlinkFeature()
3044
class _HardlinkFeature(Feature):
3047
return osutils.has_hardlinks()
3049
def feature_name(self):
3052
HardlinkFeature = _HardlinkFeature()
3055
class _OsFifoFeature(Feature):
3058
return getattr(os, 'mkfifo', None)
3060
def feature_name(self):
3061
return 'filesystem fifos'
3063
OsFifoFeature = _OsFifoFeature()
3066
class _UnicodeFilenameFeature(Feature):
3067
"""Does the filesystem support Unicode filenames?"""
3072
except UnicodeEncodeError:
3074
except (IOError, OSError):
3075
# The filesystem allows the Unicode filename but the file doesn't
3079
# The filesystem allows the Unicode filename and the file exists,
3083
UnicodeFilenameFeature = _UnicodeFilenameFeature()
3086
class TestScenarioApplier(object):
3087
"""A tool to apply scenarios to tests."""
3089
def adapt(self, test):
3090
"""Return a TestSuite containing a copy of test for each scenario."""
3091
result = unittest.TestSuite()
3092
for scenario in self.scenarios:
3093
result.addTest(self.adapt_test_to_scenario(test, scenario))
3096
def adapt_test_to_scenario(self, test, scenario):
3097
"""Copy test and apply scenario to it.
3099
:param test: A test to adapt.
3100
:param scenario: A tuple describing the scenarion.
3101
The first element of the tuple is the new test id.
3102
The second element is a dict containing attributes to set on the
3104
:return: The adapted test.
3106
from copy import deepcopy
3107
new_test = deepcopy(test)
3108
for name, value in scenario[1].items():
3109
setattr(new_test, name, value)
3110
new_id = "%s(%s)" % (new_test.id(), scenario[0])
3111
new_test.id = lambda: new_id
3115
def probe_unicode_in_user_encoding():
3116
"""Try to encode several unicode strings to use in unicode-aware tests.
3117
Return first successfull match.
3119
:return: (unicode value, encoded plain string value) or (None, None)
3121
possible_vals = [u'm\xb5', u'\xe1', u'\u0410']
3122
for uni_val in possible_vals:
3124
str_val = uni_val.encode(bzrlib.user_encoding)
3125
except UnicodeEncodeError:
3126
# Try a different character
3129
return uni_val, str_val
3133
def probe_bad_non_ascii(encoding):
3134
"""Try to find [bad] character with code [128..255]
3135
that cannot be decoded to unicode in some encoding.
3136
Return None if all non-ascii characters is valid
3139
for i in xrange(128, 256):
3142
char.decode(encoding)
3143
except UnicodeDecodeError:
3148
class _FTPServerFeature(Feature):
3149
"""Some tests want an FTP Server, check if one is available.
3151
Right now, the only way this is available is if 'medusa' is installed.
3152
http://www.amk.ca/python/code/medusa.html
3157
import bzrlib.tests.ftp_server
3162
def feature_name(self):
3165
FTPServerFeature = _FTPServerFeature()
3168
class _CaseInsensitiveFilesystemFeature(Feature):
3169
"""Check if underlined filesystem is case-insensitive
3170
(e.g. on Windows, Cygwin, MacOS)
3174
if TestCaseWithMemoryTransport.TEST_ROOT is None:
3175
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
3176
TestCaseWithMemoryTransport.TEST_ROOT = root
3178
root = TestCaseWithMemoryTransport.TEST_ROOT
3179
tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
3181
name_a = osutils.pathjoin(tdir, 'a')
3182
name_A = osutils.pathjoin(tdir, 'A')
3184
result = osutils.isdir(name_A)
3185
_rmtree_temp_dir(tdir)
3188
def feature_name(self):
3189
return 'case-insensitive filesystem'
3191
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
337
for m in (bzrlib.store, bzrlib.inventory, bzrlib.branch,
338
bzrlib.osutils, bzrlib.commands, bzrlib.merge3):
339
if m not in MODULES_TO_DOCTEST:
340
MODULES_TO_DOCTEST.append(m)
342
TestCase.BZRPATH = os.path.join(os.path.realpath(os.path.dirname(bzrlib.__path__[0])), 'bzr')
343
print '%-30s %s' % ('bzr binary', TestCase.BZRPATH)
346
suite.addTest(TestLoader().loadTestsFromNames(testmod_names))
347
for m in MODULES_TO_TEST:
348
suite.addTest(TestLoader().loadTestsFromModule(m))
349
for m in (MODULES_TO_DOCTEST):
350
suite.addTest(DocTestSuite(m))
351
for p in bzrlib.plugin.all_plugins:
352
if hasattr(p, 'test_suite'):
353
suite.addTest(p.test_suite())
added
removed
bzrlib/selftest/__init__.py
