50
54
# nb: check this before importing anything else from within it
51
55
_testtools_version = getattr(testtools, '__version__', ())
52
if _testtools_version < (0, 9, 5):
53
raise ImportError("need at least testtools 0.9.5: %s is %r"
56
if _testtools_version < (0, 9, 2):
57
raise ImportError("need at least testtools 0.9.2: %s is %r"
54
58
% (testtools.__file__, _testtools_version))
55
59
from testtools import content
58
61
from bzrlib import (
62
commands as _mod_commands,
71
plugin as _mod_plugin,
78
transport as _mod_transport,
79
import bzrlib.commands
80
import bzrlib.timestamp
82
import bzrlib.inventory
83
import bzrlib.iterablefile
82
86
import bzrlib.lsprof
83
87
except ImportError:
84
88
# lsprof not available
86
from bzrlib.smart import client, request
87
from bzrlib.transport import (
90
from bzrlib.merge import merge_inner
93
from bzrlib.smart import client, request, server
95
from bzrlib import symbol_versioning
91
96
from bzrlib.symbol_versioning import (
92
98
deprecated_function,
95
from bzrlib.tests import (
104
from bzrlib.transport import get_transport, pathfilter
105
import bzrlib.transport
106
from bzrlib.transport.local import LocalURLServer
107
from bzrlib.transport.memory import MemoryServer
108
from bzrlib.transport.readonly import ReadonlyServer
109
from bzrlib.trace import mutter, note
110
from bzrlib.tests import TestUtil
111
from bzrlib.tests.http_server import HttpServer
112
from bzrlib.tests.TestUtil import (
116
from bzrlib.tests.treeshape import build_tree_contents
100
117
from bzrlib.ui import NullProgressView
101
118
from bzrlib.ui.text import TextUIFactory
119
import bzrlib.version_info_formats.format_custom
120
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
103
122
# Mark this python module as being part of the implementation
104
123
# of unittest: this gives us better tracebacks where the last
105
124
# shown frame is the test code, not our assertXYZ.
108
default_transport = test_server.LocalURLServer
127
default_transport = LocalURLServer
111
130
_unitialized_attr = object()
116
135
SUBUNIT_SEEK_SET = 0
117
136
SUBUNIT_SEEK_CUR = 1
119
# These are intentionally brought into this namespace. That way plugins, etc
120
# can just "from bzrlib.tests import TestCase, TestLoader, etc"
121
TestSuite = TestUtil.TestSuite
122
TestLoader = TestUtil.TestLoader
124
# Tests should run in a clean and clearly defined environment. The goal is to
125
# keep them isolated from the running environment as mush as possible. The test
126
# framework ensures the variables defined below are set (or deleted if the
127
# value is None) before a test is run and reset to their original value after
128
# the test is run. Generally if some code depends on an environment variable,
129
# the tests should start without this variable in the environment. There are a
130
# few exceptions but you shouldn't violate this rule lightly.
134
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
135
# tests do check our impls match APPDATA
136
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
140
'BZREMAIL': None, # may still be present in the environment
141
'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
142
'BZR_PROGRESS_BAR': None,
143
# This should trap leaks to ~/.bzr.log. This occurs when tests use TestCase
144
# as a base class instead of TestCaseInTempDir. Tests inheriting from
145
# TestCase should not use disk resources, BZR_LOG is one.
146
'BZR_LOG': '/you-should-use-TestCaseInTempDir-if-you-need-a-log-file',
147
'BZR_PLUGIN_PATH': None,
148
'BZR_DISABLE_PLUGINS': None,
149
'BZR_PLUGINS_AT': None,
150
'BZR_CONCURRENCY': None,
151
# Make sure that any text ui tests are consistent regardless of
152
# the environment the test case is run in; you may want tests that
153
# test other combinations. 'dumb' is a reasonable guess for tests
154
# going to a pipe or a StringIO.
160
'SSH_AUTH_SOCK': None,
170
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
171
# least. If you do (care), please update this comment
175
'BZR_REMOTE_PATH': None,
176
# Generally speaking, we don't want apport reporting on crashes in
177
# the test envirnoment unless we're specifically testing apport,
178
# so that it doesn't leak into the real system environment. We
179
# use an env var so it propagates to subprocesses.
180
'APPORT_DISABLE': '1',
184
def override_os_environ(test, env=None):
185
"""Modify os.environ keeping a copy.
187
:param test: A test instance
189
:param env: A dict containing variable definitions to be installed
192
env = isolated_environ
193
test._original_os_environ = dict([(var, value)
194
for var, value in os.environ.iteritems()])
195
for var, value in env.iteritems():
196
osutils.set_or_unset_env(var, value)
197
if var not in test._original_os_environ:
198
# The var is new, add it with a value of None, so
199
# restore_os_environ will delete it
200
test._original_os_environ[var] = None
203
def restore_os_environ(test):
204
"""Restore os.environ to its original state.
206
:param test: A test instance previously passed to override_os_environ.
208
for var, value in test._original_os_environ.iteritems():
209
# Restore the original value (or delete it if the value has been set to
210
# None in override_os_environ).
211
osutils.set_or_unset_env(var, value)
214
class ExtendedTestResult(testtools.TextTestResult):
139
class ExtendedTestResult(unittest._TextTestResult):
215
140
"""Accepts, reports and accumulates the results of running tests.
217
142
Compared to the unittest version this class adds support for
266
191
self._overall_start_time = time.time()
267
192
self._strict = strict
268
self._first_thread_leaker_id = None
269
self._tests_leaking_threads_count = 0
270
self._traceback_from_test = None
272
194
def stopTestRun(self):
273
195
run = self.testsRun
274
196
actionTaken = "Ran"
275
197
stopTime = time.time()
276
198
timeTaken = stopTime - self.startTime
277
# GZ 2010-07-19: Seems testtools has no printErrors method, and though
278
# the parent class method is similar have to duplicate
279
self._show_list('ERROR', self.errors)
280
self._show_list('FAIL', self.failures)
281
self.stream.write(self.sep2)
282
self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,
200
self.stream.writeln(self.separator2)
201
self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
283
202
run, run != 1 and "s" or "", timeTaken))
203
self.stream.writeln()
284
204
if not self.wasSuccessful():
285
205
self.stream.write("FAILED (")
286
206
failed, errored = map(len, (self.failures, self.errors))
293
213
if failed or errored: self.stream.write(", ")
294
214
self.stream.write("known_failure_count=%d" %
295
215
self.known_failure_count)
296
self.stream.write(")\n")
216
self.stream.writeln(")")
298
218
if self.known_failure_count:
299
self.stream.write("OK (known_failures=%d)\n" %
219
self.stream.writeln("OK (known_failures=%d)" %
300
220
self.known_failure_count)
302
self.stream.write("OK\n")
222
self.stream.writeln("OK")
303
223
if self.skip_count > 0:
304
224
skipped = self.skip_count
305
self.stream.write('%d test%s skipped\n' %
225
self.stream.writeln('%d test%s skipped' %
306
226
(skipped, skipped != 1 and "s" or ""))
307
227
if self.unsupported:
308
228
for feature, count in sorted(self.unsupported.items()):
309
self.stream.write("Missing feature '%s' skipped %d tests.\n" %
229
self.stream.writeln("Missing feature '%s' skipped %d tests." %
310
230
(feature, count))
312
232
ok = self.wasStrictlySuccessful()
314
234
ok = self.wasSuccessful()
315
if self._first_thread_leaker_id:
235
if TestCase._first_thread_leaker_id:
316
236
self.stream.write(
317
237
'%s is leaking threads among %d leaking tests.\n' % (
318
self._first_thread_leaker_id,
319
self._tests_leaking_threads_count))
238
TestCase._first_thread_leaker_id,
239
TestCase._leaking_threads_tests))
320
240
# We don't report the main thread as an active one.
321
241
self.stream.write(
322
242
'%d non-main threads were left active in the end.\n'
323
% (len(self._active_threads) - 1))
243
% (TestCase._active_threads - 1))
325
245
def getDescription(self, test):
352
271
def _shortened_test_description(self, test):
354
what = re.sub(r'^bzrlib\.tests\.', '', what)
273
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
357
# GZ 2010-10-04: Cloned tests may end up harmlessly calling this method
358
# multiple times in a row, because the handler is added for
359
# each test but the container list is shared between cases.
360
# See lp:498869 lp:625574 and lp:637725 for background.
361
def _record_traceback_from_test(self, exc_info):
362
"""Store the traceback from passed exc_info tuple till"""
363
self._traceback_from_test = exc_info[2]
365
276
def startTest(self, test):
366
super(ExtendedTestResult, self).startTest(test)
277
unittest.TestResult.startTest(self, test)
367
278
if self.count == 0:
368
279
self.startTests()
370
280
self.report_test_start(test)
371
281
test.number = self.count
372
282
self._recordTestStartTime()
373
# Make testtools cases give us the real traceback on failure
374
addOnException = getattr(test, "addOnException", None)
375
if addOnException is not None:
376
addOnException(self._record_traceback_from_test)
377
# Only check for thread leaks on bzrlib derived test cases
378
if isinstance(test, TestCase):
379
test.addCleanup(self._check_leaked_threads, test)
381
def stopTest(self, test):
382
super(ExtendedTestResult, self).stopTest(test)
383
# Manually break cycles, means touching various private things but hey
384
getDetails = getattr(test, "getDetails", None)
385
if getDetails is not None:
387
# Clear _type_equality_funcs to try to stop TestCase instances
388
# from wasting memory. 'clear' is not available in all Python
389
# versions (bug 809048)
390
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
391
if type_equality_funcs is not None:
392
tef_clear = getattr(type_equality_funcs, "clear", None)
393
if tef_clear is None:
394
tef_instance_dict = getattr(type_equality_funcs, "__dict__", None)
395
if tef_instance_dict is not None:
396
tef_clear = tef_instance_dict.clear
397
if tef_clear is not None:
399
self._traceback_from_test = None
401
284
def startTests(self):
402
self.report_tests_starting()
403
self._active_threads = threading.enumerate()
405
def _check_leaked_threads(self, test):
406
"""See if any threads have leaked since last call
408
A sample of live threads is stored in the _active_threads attribute,
409
when this method runs it compares the current live threads and any not
410
in the previous sample are treated as having leaked.
412
now_active_threads = set(threading.enumerate())
413
threads_leaked = now_active_threads.difference(self._active_threads)
415
self._report_thread_leak(test, threads_leaked, now_active_threads)
416
self._tests_leaking_threads_count += 1
417
if self._first_thread_leaker_id is None:
418
self._first_thread_leaker_id = test.id()
419
self._active_threads = now_active_threads
286
if getattr(sys, 'frozen', None) is None:
287
bzr_path = osutils.realpath(sys.argv[0])
289
bzr_path = sys.executable
291
'bzr selftest: %s\n' % (bzr_path,))
294
bzrlib.__path__[0],))
296
' bzr-%s python-%s %s\n' % (
297
bzrlib.version_string,
298
bzrlib._format_version_tuple(sys.version_info),
299
platform.platform(aliased=1),
301
self.stream.write('\n')
421
303
def _recordTestStartTime(self):
422
304
"""Record that a test has started."""
423
self._start_datetime = self._now()
305
self._start_time = time.time()
307
def _cleanupLogFile(self, test):
308
# We can only do this if we have one of our TestCases, not if
310
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
311
if setKeepLogfile is not None:
425
314
def addError(self, test, err):
426
315
"""Tell result that test finished with an error.
460
351
self._formatTime(benchmark_time),
462
353
self.report_success(test)
463
super(ExtendedTestResult, self).addSuccess(test)
354
self._cleanupLogFile(test)
355
unittest.TestResult.addSuccess(self, test)
464
356
test._log_contents = ''
466
358
def addExpectedFailure(self, test, err):
467
359
self.known_failure_count += 1
468
360
self.report_known_failure(test, err)
470
def addUnexpectedSuccess(self, test, details=None):
471
"""Tell result the test unexpectedly passed, counting as a failure
473
When the minimum version of testtools required becomes 0.9.8 this
474
can be updated to use the new handling there.
476
super(ExtendedTestResult, self).addFailure(test, details=details)
477
self.failure_count += 1
478
self.report_unexpected_success(test,
479
"".join(details["reason"].iter_text()))
483
362
def addNotSupported(self, test, feature):
484
363
"""The test will not be run because of a missing feature.
518
396
raise errors.BzrError("Unknown whence %r" % whence)
520
def report_tests_starting(self):
521
"""Display information before the test run begins"""
522
if getattr(sys, 'frozen', None) is None:
523
bzr_path = osutils.realpath(sys.argv[0])
525
bzr_path = sys.executable
527
'bzr selftest: %s\n' % (bzr_path,))
530
bzrlib.__path__[0],))
532
' bzr-%s python-%s %s\n' % (
533
bzrlib.version_string,
534
bzrlib._format_version_tuple(sys.version_info),
535
platform.platform(aliased=1),
537
self.stream.write('\n')
539
def report_test_start(self, test):
540
"""Display information on the test just about to be run"""
542
def _report_thread_leak(self, test, leaked_threads, active_threads):
543
"""Display information on a test that leaked one or more threads"""
544
# GZ 2010-09-09: A leak summary reported separately from the general
545
# thread debugging would be nice. Tests under subunit
546
# need something not using stream, perhaps adding a
547
# testtools details object would be fitting.
548
if 'threads' in selftest_debug_flags:
549
self.stream.write('%s is leaking, active is now %d\n' %
550
(test.id(), len(active_threads)))
398
def report_cleaning_up(self):
552
401
def startTestRun(self):
553
402
self.startTime = time.time()
694
546
return '%s%s' % (indent, err[1])
696
548
def report_error(self, test, err):
697
self.stream.write('ERROR %s\n%s\n'
549
self.stream.writeln('ERROR %s\n%s'
698
550
% (self._testTimeString(test),
699
551
self._error_summary(err)))
701
553
def report_failure(self, test, err):
702
self.stream.write(' FAIL %s\n%s\n'
554
self.stream.writeln(' FAIL %s\n%s'
703
555
% (self._testTimeString(test),
704
556
self._error_summary(err)))
706
558
def report_known_failure(self, test, err):
707
self.stream.write('XFAIL %s\n%s\n'
559
self.stream.writeln('XFAIL %s\n%s'
708
560
% (self._testTimeString(test),
709
561
self._error_summary(err)))
711
def report_unexpected_success(self, test, reason):
712
self.stream.write(' FAIL %s\n%s: %s\n'
713
% (self._testTimeString(test),
714
"Unexpected success. Should have failed",
717
563
def report_success(self, test):
718
self.stream.write(' OK %s\n' % self._testTimeString(test))
564
self.stream.writeln(' OK %s' % self._testTimeString(test))
719
565
for bench_called, stats in getattr(test, '_benchcalls', []):
720
self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
566
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
721
567
stats.pprint(file=self.stream)
722
568
# flush the stream so that we get smooth output. This verbose mode is
723
569
# used to show the output in PQM.
724
570
self.stream.flush()
726
572
def report_skip(self, test, reason):
727
self.stream.write(' SKIP %s\n%s\n'
573
self.stream.writeln(' SKIP %s\n%s'
728
574
% (self._testTimeString(test), reason))
730
576
def report_not_applicable(self, test, reason):
731
self.stream.write(' N/A %s\n %s\n'
577
self.stream.writeln(' N/A %s\n %s'
732
578
% (self._testTimeString(test), reason))
734
580
def report_unsupported(self, test, feature):
735
581
"""test cannot be run because feature is missing."""
736
self.stream.write("NODEP %s\n The feature '%s' is not available.\n"
582
self.stream.writeln("NODEP %s\n The feature '%s' is not available."
737
583
%(self._testTimeString(test), feature))
960
788
routine, and to build and check bzr trees.
962
790
In addition to the usual method of overriding tearDown(), this class also
963
allows subclasses to register cleanup functions via addCleanup, which are
791
allows subclasses to register functions into the _cleanups list, which is
964
792
run in order as the object is torn down. It's less likely this will be
965
793
accidentally overlooked.
796
_active_threads = None
797
_leaking_threads_tests = 0
798
_first_thread_leaker_id = None
799
_log_file_name = None
969
800
# record lsprof data when performing benchmark calls.
970
801
_gather_lsprof_in_benchmarks = False
972
803
def __init__(self, methodName='testMethod'):
973
804
super(TestCase, self).__init__(methodName)
974
806
self._directory_isolation = True
975
807
self.exception_handlers.insert(0,
976
808
(UnavailableFeature, self._do_unsupported_or_skip))
990
826
self._track_transports()
991
827
self._track_locks()
992
828
self._clear_debug_flags()
993
# Isolate global verbosity level, to make sure it's reproducible
994
# between tests. We should get rid of this altogether: bug 656694. --
996
self.overrideAttr(bzrlib.trace, '_verbosity_level', 0)
997
# Isolate config option expansion until its default value for bzrlib is
998
# settled on or a the FIXME associated with _get_expand_default_value
999
# is addressed -- vila 20110219
1000
self.overrideAttr(config, '_expand_default_value', None)
1001
self._log_files = set()
1002
# Each key in the ``_counters`` dict holds a value for a different
1003
# counter. When the test ends, addDetail() should be used to output the
1004
# counter values. This happens in install_counter_hook().
1006
if 'config_stats' in selftest_debug_flags:
1007
self._install_config_stats_hooks()
829
TestCase._active_threads = threading.activeCount()
830
self.addCleanup(self._check_leaked_threads)
1009
832
def debug(self):
1010
833
# debug a frame up.
1012
835
pdb.Pdb().set_trace(sys._getframe().f_back)
1014
def discardDetail(self, name):
1015
"""Extend the addDetail, getDetails api so we can remove a detail.
1017
eg. bzr always adds the 'log' detail at startup, but we don't want to
1018
include it for skipped, xfail, etc tests.
1020
It is safe to call this for a detail that doesn't exist, in case this
1021
gets called multiple times.
1023
# We cheat. details is stored in __details which means we shouldn't
1024
# touch it. but getDetails() returns the dict directly, so we can
1026
details = self.getDetails()
1030
def install_counter_hook(self, hooks, name, counter_name=None):
1031
"""Install a counting hook.
1033
Any hook can be counted as long as it doesn't need to return a value.
1035
:param hooks: Where the hook should be installed.
1037
:param name: The hook name that will be counted.
1039
:param counter_name: The counter identifier in ``_counters``, defaults
1042
_counters = self._counters # Avoid closing over self
1043
if counter_name is None:
1045
if _counters.has_key(counter_name):
1046
raise AssertionError('%s is already used as a counter name'
1048
_counters[counter_name] = 0
1049
self.addDetail(counter_name, content.Content(content.UTF8_TEXT,
1050
lambda: ['%d' % (_counters[counter_name],)]))
1051
def increment_counter(*args, **kwargs):
1052
_counters[counter_name] += 1
1053
label = 'count %s calls' % (counter_name,)
1054
hooks.install_named_hook(name, increment_counter, label)
1055
self.addCleanup(hooks.uninstall_named_hook, name, label)
1057
def _install_config_stats_hooks(self):
1058
"""Install config hooks to count hook calls.
1061
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1062
self.install_counter_hook(config.ConfigHooks, hook_name,
1063
'config.%s' % (hook_name,))
1065
# The OldConfigHooks are private and need special handling to protect
1066
# against recursive tests (tests that run other tests), so we just do
1067
# manually what registering them into _builtin_known_hooks will provide
1069
self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1070
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1071
self.install_counter_hook(config.OldConfigHooks, hook_name,
1072
'old_config.%s' % (hook_name,))
837
def _check_leaked_threads(self):
838
active = threading.activeCount()
839
leaked_threads = active - TestCase._active_threads
840
TestCase._active_threads = active
841
# If some tests make the number of threads *decrease*, we'll consider
842
# that they are just observing old threads dieing, not agressively kill
843
# random threads. So we don't report these tests as leaking. The risk
844
# is that we have false positives that way (the test see 2 threads
845
# going away but leak one) but it seems less likely than the actual
846
# false positives (the test see threads going away and does not leak).
847
if leaked_threads > 0:
848
TestCase._leaking_threads_tests += 1
849
if TestCase._first_thread_leaker_id is None:
850
TestCase._first_thread_leaker_id = self.id()
1074
852
def _clear_debug_flags(self):
1075
853
"""Prevent externally set debug flags affecting tests.
1087
865
def _clear_hooks(self):
1088
866
# prevent hooks affecting tests
1089
known_hooks = hooks.known_hooks
1090
867
self._preserved_hooks = {}
1091
for key, (parent, name) in known_hooks.iter_parent_objects():
1092
current_hooks = getattr(parent, name)
868
for key, factory in hooks.known_hooks.items():
869
parent, name = hooks.known_hooks_key_to_parent_and_attribute(key)
870
current_hooks = hooks.known_hooks_key_to_object(key)
1093
871
self._preserved_hooks[parent] = (name, current_hooks)
1094
self._preserved_lazy_hooks = hooks._lazy_hooks
1095
hooks._lazy_hooks = {}
1096
872
self.addCleanup(self._restoreHooks)
1097
for key, (parent, name) in known_hooks.iter_parent_objects():
1098
factory = known_hooks.get(key)
873
for key, factory in hooks.known_hooks.items():
874
parent, name = hooks.known_hooks_key_to_parent_and_attribute(key)
1099
875
setattr(parent, name, factory())
1100
876
# this hook should always be installed
1101
877
request._install_hook()
1362
1133
'st_mtime did not match')
1363
1134
self.assertEqual(expected.st_ctime, actual.st_ctime,
1364
1135
'st_ctime did not match')
1365
if sys.platform == 'win32':
1136
if sys.platform != 'win32':
1366
1137
# On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
1367
1138
# is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
1368
# odd. We just force it to always be 0 to avoid any problems.
1369
self.assertEqual(0, expected.st_dev)
1370
self.assertEqual(0, actual.st_dev)
1371
self.assertEqual(0, expected.st_ino)
1372
self.assertEqual(0, actual.st_ino)
1139
# odd. Regardless we shouldn't actually try to assert anything
1140
# about their values
1374
1141
self.assertEqual(expected.st_dev, actual.st_dev,
1375
1142
'st_dev did not match')
1376
1143
self.assertEqual(expected.st_ino, actual.st_ino,
1548
1308
self.assertEqualDiff(content, s)
1550
def assertDocstring(self, expected_docstring, obj):
1551
"""Fail if obj does not have expected_docstring"""
1553
# With -OO the docstring should be None instead
1554
self.assertIs(obj.__doc__, None)
1556
self.assertEqual(expected_docstring, obj.__doc__)
1558
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1559
1310
def failUnlessExists(self, path):
1560
return self.assertPathExists(path)
1562
def assertPathExists(self, path):
1563
1311
"""Fail unless path or paths, which may be abs or relative, exist."""
1564
1312
if not isinstance(path, basestring):
1566
self.assertPathExists(p)
1314
self.failUnlessExists(p)
1568
self.assertTrue(osutils.lexists(path),
1569
path + " does not exist")
1316
self.failUnless(osutils.lexists(path),path+" does not exist")
1571
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1572
1318
def failIfExists(self, path):
1573
return self.assertPathDoesNotExist(path)
1575
def assertPathDoesNotExist(self, path):
1576
1319
"""Fail if path or paths, which may be abs or relative, exist."""
1577
1320
if not isinstance(path, basestring):
1579
self.assertPathDoesNotExist(p)
1322
self.failIfExists(p)
1581
self.assertFalse(osutils.lexists(path),
1324
self.failIf(osutils.lexists(path),path+" exists")
1584
1326
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1585
1327
"""A helper for callDeprecated and applyDeprecated.
1701
1442
The file is removed as the test is torn down.
1703
pseudo_log_file = StringIO()
1704
def _get_log_contents_for_weird_testtools_api():
1705
return [pseudo_log_file.getvalue().decode(
1706
"utf-8", "replace").encode("utf-8")]
1707
self.addDetail("log", content.Content(content.ContentType("text",
1708
"plain", {"charset": "utf8"}),
1709
_get_log_contents_for_weird_testtools_api))
1710
self._log_file = pseudo_log_file
1711
self._log_memento = trace.push_log_file(self._log_file)
1444
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
1445
self._log_file = os.fdopen(fileno, 'w+')
1446
self._log_memento = bzrlib.trace.push_log_file(self._log_file)
1447
self._log_file_name = name
1712
1448
self.addCleanup(self._finishLogFile)
1714
1450
def _finishLogFile(self):
1715
1451
"""Finished with the log file.
1717
Close the file and delete it.
1453
Close the file and delete it, unless setKeepLogfile was called.
1719
if trace._trace_file:
1455
if bzrlib.trace._trace_file:
1720
1456
# flush the log file, to get all content
1721
trace._trace_file.flush()
1722
trace.pop_log_file(self._log_memento)
1457
bzrlib.trace._trace_file.flush()
1458
bzrlib.trace.pop_log_file(self._log_memento)
1459
# Cache the log result and delete the file on disk
1460
self._get_log(False)
1724
1462
def thisFailsStrictLockCheck(self):
1725
1463
"""It is known that this test would fail with -Dstrict_locks.
1756
1499
setattr(obj, attr_name, new)
1759
def overrideEnv(self, name, new):
1760
"""Set an environment variable, and reset it after the test.
1762
:param name: The environment variable name.
1764
:param new: The value to set the variable to. If None, the
1765
variable is deleted from the environment.
1767
:returns: The actual variable value.
1769
value = osutils.set_or_unset_env(name, new)
1770
self.addCleanup(osutils.set_or_unset_env, name, value)
1773
def recordCalls(self, obj, attr_name):
1774
"""Monkeypatch in a wrapper that will record calls.
1776
The monkeypatch is automatically removed when the test concludes.
1778
:param obj: The namespace holding the reference to be replaced;
1779
typically a module, class, or object.
1780
:param attr_name: A string for the name of the attribute to
1782
:returns: A list that will be extended with one item every time the
1783
function is called, with a tuple of (args, kwargs).
1787
def decorator(*args, **kwargs):
1788
calls.append((args, kwargs))
1789
return orig(*args, **kwargs)
1790
orig = self.overrideAttr(obj, attr_name, decorator)
1793
1502
def _cleanEnvironment(self):
1794
for name, value in isolated_environ.iteritems():
1795
self.overrideEnv(name, value)
1504
'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
1505
'HOME': os.getcwd(),
1506
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
1507
# tests do check our impls match APPDATA
1508
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
1512
'BZREMAIL': None, # may still be present in the environment
1514
'BZR_PROGRESS_BAR': None,
1516
'BZR_PLUGIN_PATH': None,
1517
'BZR_CONCURRENCY': None,
1518
# Make sure that any text ui tests are consistent regardless of
1519
# the environment the test case is run in; you may want tests that
1520
# test other combinations. 'dumb' is a reasonable guess for tests
1521
# going to a pipe or a StringIO.
1525
'BZR_COLUMNS': '80',
1527
'SSH_AUTH_SOCK': None,
1531
'https_proxy': None,
1532
'HTTPS_PROXY': None,
1537
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
1538
# least. If you do (care), please update this comment
1542
'BZR_REMOTE_PATH': None,
1543
# Generally speaking, we don't want apport reporting on crashes in
1544
# the test envirnoment unless we're specifically testing apport,
1545
# so that it doesn't leak into the real system environment. We
1546
# use an env var so it propagates to subprocesses.
1547
'APPORT_DISABLE': '1',
1550
self.addCleanup(self._restoreEnvironment)
1551
for name, value in new_env.iteritems():
1552
self._captureVar(name, value)
1554
def _captureVar(self, name, newvalue):
1555
"""Set an environment variable, and reset it when finished."""
1556
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
1558
def _restoreEnvironment(self):
1559
for name, value in self.__old_env.iteritems():
1560
osutils.set_or_unset_env(name, value)
1797
1562
def _restoreHooks(self):
1798
1563
for klass, (name, hooks) in self._preserved_hooks.items():
1799
1564
setattr(klass, name, hooks)
1800
self._preserved_hooks.clear()
1801
bzrlib.hooks._lazy_hooks = self._preserved_lazy_hooks
1802
self._preserved_lazy_hooks.clear()
1804
1566
def knownFailure(self, reason):
1805
"""Declare that this test fails for a known reason
1807
Tests that are known to fail should generally be using expectedFailure
1808
with an appropriate reverse assertion if a change could cause the test
1809
to start passing. Conversely if the test has no immediate prospect of
1810
succeeding then using skip is more suitable.
1812
When this method is called while an exception is being handled, that
1813
traceback will be used, otherwise a new exception will be thrown to
1814
provide one but won't be reported.
1816
self._add_reason(reason)
1818
exc_info = sys.exc_info()
1819
if exc_info != (None, None, None):
1820
self._report_traceback(exc_info)
1823
raise self.failureException(reason)
1824
except self.failureException:
1825
exc_info = sys.exc_info()
1826
# GZ 02-08-2011: Maybe cleanup this err.exc_info attribute too?
1827
raise testtools.testcase._ExpectedFailure(exc_info)
1831
def _suppress_log(self):
1832
"""Remove the log info from details."""
1833
self.discardDetail('log')
1567
"""This test has failed for some known reason."""
1568
raise KnownFailure(reason)
1835
1570
def _do_skip(self, result, reason):
1836
self._suppress_log()
1837
1571
addSkip = getattr(result, 'addSkip', None)
1838
1572
if not callable(addSkip):
1839
1573
result.addSuccess(result)
1864
1596
self._do_skip(result, reason)
1867
def _report_skip(self, result, err):
1868
"""Override the default _report_skip.
1870
We want to strip the 'log' detail. If we waint until _do_skip, it has
1871
already been formatted into the 'reason' string, and we can't pull it
1874
self._suppress_log()
1875
super(TestCase, self)._report_skip(self, result, err)
1878
def _report_expected_failure(self, result, err):
1881
See _report_skip for motivation.
1883
self._suppress_log()
1884
super(TestCase, self)._report_expected_failure(self, result, err)
1887
1599
def _do_unsupported_or_skip(self, result, e):
1888
1600
reason = e.args[0]
1889
self._suppress_log()
1890
1601
addNotSupported = getattr(result, 'addNotSupported', None)
1891
1602
if addNotSupported is not None:
1892
1603
result.addNotSupported(self, reason)
1918
1629
self._benchtime += time.time() - start
1920
1631
def log(self, *args):
1634
def _get_log(self, keep_log_file=False):
1635
"""Internal helper to get the log from bzrlib.trace for this test.
1637
Please use self.getDetails, or self.get_log to access this in test case
1640
:param keep_log_file: When True, if the log is still a file on disk
1641
leave it as a file on disk. When False, if the log is still a file
1642
on disk, the log file is deleted and the log preserved as
1644
:return: A string containing the log.
1646
if self._log_contents is not None:
1648
self._log_contents.decode('utf8')
1649
except UnicodeDecodeError:
1650
unicodestr = self._log_contents.decode('utf8', 'replace')
1651
self._log_contents = unicodestr.encode('utf8')
1652
return self._log_contents
1654
if bzrlib.trace._trace_file:
1655
# flush the log file, to get all content
1656
bzrlib.trace._trace_file.flush()
1657
if self._log_file_name is not None:
1658
logfile = open(self._log_file_name)
1660
log_contents = logfile.read()
1664
log_contents.decode('utf8')
1665
except UnicodeDecodeError:
1666
unicodestr = log_contents.decode('utf8', 'replace')
1667
log_contents = unicodestr.encode('utf8')
1668
if not keep_log_file:
1669
self._log_file.close()
1670
self._log_file = None
1671
# Permit multiple calls to get_log until we clean it up in
1673
self._log_contents = log_contents
1675
os.remove(self._log_file_name)
1677
if sys.platform == 'win32' and e.errno == errno.EACCES:
1678
sys.stderr.write(('Unable to delete log file '
1679
' %r\n' % self._log_file_name))
1682
self._log_file_name = None
1685
return "No log file content and no log file name."
1923
1687
def get_log(self):
1924
1688
"""Get a unicode string containing the log from bzrlib.trace.
2200
def _add_subprocess_log(self, log_file_path):
2201
if len(self._log_files) == 0:
2202
# Register an addCleanup func. We do this on the first call to
2203
# _add_subprocess_log rather than in TestCase.setUp so that this
2204
# addCleanup is registered after any cleanups for tempdirs that
2205
# subclasses might create, which will probably remove the log file
2207
self.addCleanup(self._subprocess_log_cleanup)
2208
# self._log_files is a set, so if a log file is reused we won't grab it
2210
self._log_files.add(log_file_path)
2212
def _subprocess_log_cleanup(self):
2213
for count, log_file_path in enumerate(self._log_files):
2214
# We use buffer_now=True to avoid holding the file open beyond
2215
# the life of this function, which might interfere with e.g.
2216
# cleaning tempdirs on Windows.
2217
# XXX: Testtools 0.9.5 doesn't have the content_from_file helper
2218
#detail_content = content.content_from_file(
2219
# log_file_path, buffer_now=True)
2220
with open(log_file_path, 'rb') as log_file:
2221
log_file_bytes = log_file.read()
2222
detail_content = content.Content(content.ContentType("text",
2223
"plain", {"charset": "utf8"}), lambda: [log_file_bytes])
2224
self.addDetail("start_bzr_subprocess-log-%d" % (count,),
2227
1955
def _popen(self, *args, **kwargs):
2228
1956
"""Place a call to Popen.
2230
1958
Allows tests to override this method to intercept the calls made to
2231
1959
Popen for introspection.
2233
return subprocess.Popen(*args, **kwargs)
1961
return Popen(*args, **kwargs)
2235
1963
def get_source_path(self):
2236
1964
"""Return the path of the directory containing bzrlib."""
2266
1994
if retcode is not None and retcode != process.returncode:
2267
1995
if process_args is None:
2268
1996
process_args = "(unknown args)"
2269
trace.mutter('Output of bzr %s:\n%s', process_args, out)
2270
trace.mutter('Error for bzr %s:\n%s', process_args, err)
1997
mutter('Output of bzr %s:\n%s', process_args, out)
1998
mutter('Error for bzr %s:\n%s', process_args, err)
2271
1999
self.fail('Command bzr %s failed with retcode %s != %s'
2272
2000
% (process_args, retcode, process.returncode))
2273
2001
return [out, err]
2275
def check_tree_shape(self, tree, shape):
2276
"""Compare a tree to a list of expected names.
2003
def check_inventory_shape(self, inv, shape):
2004
"""Compare an inventory to a list of expected names.
2278
2006
Fail if they are not precisely equal.
2281
2009
shape = list(shape) # copy
2282
for path, ie in tree.iter_entries_by_dir():
2010
for path, ie in inv.entries():
2283
2011
name = path.replace('\\', '/')
2284
2012
if ie.kind == 'directory':
2285
2013
name = name + '/'
2287
pass # ignore root entry
2289
2015
shape.remove(name)
2291
2017
extras.append(name)
2381
2107
class TestCaseWithMemoryTransport(TestCase):
2382
2108
"""Common test class for tests that do not need disk resources.
2384
Tests that need disk resources should derive from TestCaseInTempDir
2385
orTestCaseWithTransport.
2110
Tests that need disk resources should derive from TestCaseWithTransport.
2387
2112
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
2389
For TestCaseWithMemoryTransport the ``test_home_dir`` is set to the name of
2114
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
2390
2115
a directory which does not exist. This serves to help ensure test isolation
2391
is preserved. ``test_dir`` is set to the TEST_ROOT, as is cwd, because they
2392
must exist. However, TestCaseWithMemoryTransport does not offer local file
2393
defaults for the transport in tests, nor does it obey the command line
2116
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
2117
must exist. However, TestCaseWithMemoryTransport does not offer local
2118
file defaults for the transport in tests, nor does it obey the command line
2394
2119
override, so tests that accidentally write to the common directory should
2397
:cvar TEST_ROOT: Directory containing all temporary directories, plus a
2398
``.bzr`` directory that stops us ascending higher into the filesystem.
2122
:cvar TEST_ROOT: Directory containing all temporary directories, plus
2123
a .bzr directory that stops us ascending higher into the filesystem.
2401
2126
TEST_ROOT = None
2667
2385
test_home_dir = self.test_home_dir
2668
2386
if isinstance(test_home_dir, unicode):
2669
2387
test_home_dir = test_home_dir.encode(sys.getfilesystemencoding())
2670
self.overrideEnv('HOME', test_home_dir)
2671
self.overrideEnv('BZR_HOME', test_home_dir)
2388
os.environ['HOME'] = test_home_dir
2389
os.environ['BZR_HOME'] = test_home_dir
2673
2391
def setUp(self):
2674
2392
super(TestCaseWithMemoryTransport, self).setUp()
2675
# Ensure that ConnectedTransport doesn't leak sockets
2676
def get_transport_from_url_with_cleanup(*args, **kwargs):
2677
t = orig_get_transport_from_url(*args, **kwargs)
2678
if isinstance(t, _mod_transport.ConnectedTransport):
2679
self.addCleanup(t.disconnect)
2682
orig_get_transport_from_url = self.overrideAttr(
2683
_mod_transport, 'get_transport_from_url',
2684
get_transport_from_url_with_cleanup)
2685
2393
self._make_test_root()
2686
2394
self.addCleanup(os.chdir, os.getcwdu())
2687
2395
self.makeAndChdirToTestDir()
3453
3149
def partition_tests(suite, count):
3454
3150
"""Partition suite into count lists of tests."""
3455
# This just assigns tests in a round-robin fashion. On one hand this
3456
# splits up blocks of related tests that might run faster if they shared
3457
# resources, but on the other it avoids assigning blocks of slow tests to
3458
# just one partition. So the slowest partition shouldn't be much slower
3460
partitions = [list() for i in range(count)]
3461
tests = iter_suite_tests(suite)
3462
for partition, test in itertools.izip(itertools.cycle(partitions), tests):
3463
partition.append(test)
3467
def workaround_zealous_crypto_random():
3468
"""Crypto.Random want to help us being secure, but we don't care here.
3470
This workaround some test failure related to the sftp server. Once paramiko
3471
stop using the controversial API in Crypto.Random, we may get rid of it.
3474
from Crypto.Random import atfork
3152
tests = list(iter_suite_tests(suite))
3153
tests_per_process = int(math.ceil(float(len(tests)) / count))
3154
for block in range(count):
3155
low_test = block * tests_per_process
3156
high_test = low_test + tests_per_process
3157
process_tests = tests[low_test:high_test]
3158
result.append(process_tests)
3480
3162
def fork_for_tests(suite):
3590
class ProfileResult(testtools.ExtendedToOriginalDecorator):
3269
class ForwardingResult(unittest.TestResult):
3271
def __init__(self, target):
3272
unittest.TestResult.__init__(self)
3273
self.result = target
3275
def startTest(self, test):
3276
self.result.startTest(test)
3278
def stopTest(self, test):
3279
self.result.stopTest(test)
3281
def startTestRun(self):
3282
self.result.startTestRun()
3284
def stopTestRun(self):
3285
self.result.stopTestRun()
3287
def addSkip(self, test, reason):
3288
self.result.addSkip(test, reason)
3290
def addSuccess(self, test):
3291
self.result.addSuccess(test)
3293
def addError(self, test, err):
3294
self.result.addError(test, err)
3296
def addFailure(self, test, err):
3297
self.result.addFailure(test, err)
3298
ForwardingResult = testtools.ExtendedToOriginalDecorator
3301
class ProfileResult(ForwardingResult):
3591
3302
"""Generate profiling data for all activity between start and success.
3593
3304
The profile data is appended to the test's _benchcalls attribute and can
3891
3591
'bzrlib.tests.per_repository',
3892
3592
'bzrlib.tests.per_repository_chk',
3893
3593
'bzrlib.tests.per_repository_reference',
3894
'bzrlib.tests.per_repository_vf',
3895
3594
'bzrlib.tests.per_uifactory',
3896
3595
'bzrlib.tests.per_versionedfile',
3897
3596
'bzrlib.tests.per_workingtree',
3898
3597
'bzrlib.tests.test__annotator',
3899
3598
'bzrlib.tests.test__bencode',
3900
'bzrlib.tests.test__btree_serializer',
3901
3599
'bzrlib.tests.test__chk_map',
3902
3600
'bzrlib.tests.test__dirstate_helpers',
3903
3601
'bzrlib.tests.test__groupcompress',
3925
3623
'bzrlib.tests.test_chunk_writer',
3926
3624
'bzrlib.tests.test_clean_tree',
3927
3625
'bzrlib.tests.test_cleanup',
3928
'bzrlib.tests.test_cmdline',
3929
3626
'bzrlib.tests.test_commands',
3930
3627
'bzrlib.tests.test_commit',
3931
3628
'bzrlib.tests.test_commit_merge',
3932
3629
'bzrlib.tests.test_config',
3933
3630
'bzrlib.tests.test_conflicts',
3934
'bzrlib.tests.test_controldir',
3935
3631
'bzrlib.tests.test_counted_lock',
3936
3632
'bzrlib.tests.test_crash',
3937
3633
'bzrlib.tests.test_decorators',
3938
3634
'bzrlib.tests.test_delta',
3939
3635
'bzrlib.tests.test_debug',
3636
'bzrlib.tests.test_deprecated_graph',
3940
3637
'bzrlib.tests.test_diff',
3941
3638
'bzrlib.tests.test_directory_service',
3942
3639
'bzrlib.tests.test_dirstate',
3944
3641
'bzrlib.tests.test_eol_filters',
3945
3642
'bzrlib.tests.test_errors',
3946
3643
'bzrlib.tests.test_export',
3947
'bzrlib.tests.test_export_pot',
3948
3644
'bzrlib.tests.test_extract',
3949
'bzrlib.tests.test_features',
3950
3645
'bzrlib.tests.test_fetch',
3951
'bzrlib.tests.test_fixtures',
3952
3646
'bzrlib.tests.test_fifo_cache',
3953
3647
'bzrlib.tests.test_filters',
3954
'bzrlib.tests.test_filter_tree',
3955
3648
'bzrlib.tests.test_ftp_transport',
3956
3649
'bzrlib.tests.test_foreign',
3957
3650
'bzrlib.tests.test_generate_docs',
3966
3659
'bzrlib.tests.test_http',
3967
3660
'bzrlib.tests.test_http_response',
3968
3661
'bzrlib.tests.test_https_ca_bundle',
3969
'bzrlib.tests.test_i18n',
3970
3662
'bzrlib.tests.test_identitymap',
3971
3663
'bzrlib.tests.test_ignores',
3972
3664
'bzrlib.tests.test_index',
3973
'bzrlib.tests.test_import_tariff',
3974
3665
'bzrlib.tests.test_info',
3975
3666
'bzrlib.tests.test_inv',
3976
3667
'bzrlib.tests.test_inventory_delta',
3977
3668
'bzrlib.tests.test_knit',
3978
3669
'bzrlib.tests.test_lazy_import',
3979
3670
'bzrlib.tests.test_lazy_regex',
3980
'bzrlib.tests.test_library_state',
3981
3671
'bzrlib.tests.test_lock',
3982
3672
'bzrlib.tests.test_lockable_files',
3983
3673
'bzrlib.tests.test_lockdir',
3985
3675
'bzrlib.tests.test_lru_cache',
3986
3676
'bzrlib.tests.test_lsprof',
3987
3677
'bzrlib.tests.test_mail_client',
3988
'bzrlib.tests.test_matchers',
3989
3678
'bzrlib.tests.test_memorytree',
3990
3679
'bzrlib.tests.test_merge',
3991
3680
'bzrlib.tests.test_merge3',
3992
3681
'bzrlib.tests.test_merge_core',
3993
3682
'bzrlib.tests.test_merge_directive',
3994
'bzrlib.tests.test_mergetools',
3995
3683
'bzrlib.tests.test_missing',
3996
3684
'bzrlib.tests.test_msgeditor',
3997
3685
'bzrlib.tests.test_multiparent',
4199
3875
# Some tests mentioned in the list are not in the test suite. The
4200
3876
# list may be out of date, report to the tester.
4201
3877
for id in not_found:
4202
trace.warning('"%s" not found in the test suite', id)
3878
bzrlib.trace.warning('"%s" not found in the test suite', id)
4203
3879
for id in duplicates:
4204
trace.warning('"%s" is used as an id by several tests', id)
3880
bzrlib.trace.warning('"%s" is used as an id by several tests', id)
4209
def multiply_scenarios(*scenarios):
4210
"""Multiply two or more iterables of scenarios.
4212
It is safe to pass scenario generators or iterators.
4214
:returns: A list of compound scenarios: the cross-product of all
4215
scenarios, with the names concatenated and the parameters
4218
return reduce(_multiply_two_scenarios, map(list, scenarios))
4221
def _multiply_two_scenarios(scenarios_left, scenarios_right):
3885
def multiply_scenarios(scenarios_left, scenarios_right):
4222
3886
"""Multiply two sets of scenarios.
4224
3888
:returns: the cartesian product of the two sets of scenarios, that is
4388
4039
if test_id != None:
4389
4040
ui.ui_factory.clear_term()
4390
4041
sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4391
# Ugly, but the last thing we want here is fail, so bear with it.
4392
printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4393
).encode('ascii', 'replace')
4394
4042
sys.stderr.write('Unable to remove testing dir %s\n%s'
4395
% (os.path.basename(dirname), printable_e))
4043
% (os.path.basename(dirname), e))
4046
class Feature(object):
4047
"""An operating system Feature."""
4050
self._available = None
4052
def available(self):
4053
"""Is the feature available?
4055
:return: True if the feature is available.
4057
if self._available is None:
4058
self._available = self._probe()
4059
return self._available
4062
"""Implement this method in concrete features.
4064
:return: True if the feature is available.
4066
raise NotImplementedError
4069
if getattr(self, 'feature_name', None):
4070
return self.feature_name()
4071
return self.__class__.__name__
4074
class _SymlinkFeature(Feature):
4077
return osutils.has_symlinks()
4079
def feature_name(self):
4082
SymlinkFeature = _SymlinkFeature()
4085
class _HardlinkFeature(Feature):
4088
return osutils.has_hardlinks()
4090
def feature_name(self):
4093
HardlinkFeature = _HardlinkFeature()
4096
class _OsFifoFeature(Feature):
4099
return getattr(os, 'mkfifo', None)
4101
def feature_name(self):
4102
return 'filesystem fifos'
4104
OsFifoFeature = _OsFifoFeature()
4107
class _UnicodeFilenameFeature(Feature):
4108
"""Does the filesystem support Unicode filenames?"""
4112
# Check for character combinations unlikely to be covered by any
4113
# single non-unicode encoding. We use the characters
4114
# - greek small letter alpha (U+03B1) and
4115
# - braille pattern dots-123456 (U+283F).
4116
os.stat(u'\u03b1\u283f')
4117
except UnicodeEncodeError:
4119
except (IOError, OSError):
4120
# The filesystem allows the Unicode filename but the file doesn't
4124
# The filesystem allows the Unicode filename and the file exists,
4128
UnicodeFilenameFeature = _UnicodeFilenameFeature()
4131
class _CompatabilityThunkFeature(Feature):
4132
"""This feature is just a thunk to another feature.
4134
It issues a deprecation warning if it is accessed, to let you know that you
4135
should really use a different feature.
4138
def __init__(self, dep_version, module, name,
4139
replacement_name, replacement_module=None):
4140
super(_CompatabilityThunkFeature, self).__init__()
4141
self._module = module
4142
if replacement_module is None:
4143
replacement_module = module
4144
self._replacement_module = replacement_module
4146
self._replacement_name = replacement_name
4147
self._dep_version = dep_version
4148
self._feature = None
4151
if self._feature is None:
4152
depr_msg = self._dep_version % ('%s.%s'
4153
% (self._module, self._name))
4154
use_msg = ' Use %s.%s instead.' % (self._replacement_module,
4155
self._replacement_name)
4156
symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
4157
# Import the new feature and use it as a replacement for the
4159
mod = __import__(self._replacement_module, {}, {},
4160
[self._replacement_name])
4161
self._feature = getattr(mod, self._replacement_name)
4165
return self._feature._probe()
4168
class ModuleAvailableFeature(Feature):
4169
"""This is a feature than describes a module we want to be available.
4171
Declare the name of the module in __init__(), and then after probing, the
4172
module will be available as 'self.module'.
4174
:ivar module: The module if it is available, else None.
4177
def __init__(self, module_name):
4178
super(ModuleAvailableFeature, self).__init__()
4179
self.module_name = module_name
4183
self._module = __import__(self.module_name, {}, {}, [''])
4190
if self.available(): # Make sure the probe has been done
4194
def feature_name(self):
4195
return self.module_name
4198
# This is kept here for compatibility, it is recommended to use
4199
# 'bzrlib.tests.feature.paramiko' instead
4200
ParamikoFeature = _CompatabilityThunkFeature(
4201
deprecated_in((2,1,0)),
4202
'bzrlib.tests.features', 'ParamikoFeature', 'paramiko')
4398
4205
def probe_unicode_in_user_encoding():
4238
class _HTTPSServerFeature(Feature):
4239
"""Some tests want an https Server, check if one is available.
4241
Right now, the only way this is available is under python2.6 which provides
4252
def feature_name(self):
4253
return 'HTTPSServer'
4256
HTTPSServerFeature = _HTTPSServerFeature()
4259
class _UnicodeFilename(Feature):
4260
"""Does the filesystem support Unicode filenames?"""
4265
except UnicodeEncodeError:
4267
except (IOError, OSError):
4268
# The filesystem allows the Unicode filename but the file doesn't
4272
# The filesystem allows the Unicode filename and the file exists,
4276
UnicodeFilename = _UnicodeFilename()
4279
class _UTF8Filesystem(Feature):
4280
"""Is the filesystem UTF-8?"""
4283
if osutils._fs_enc.upper() in ('UTF-8', 'UTF8'):
4287
UTF8Filesystem = _UTF8Filesystem()
4290
class _BreakinFeature(Feature):
4291
"""Does this platform support the breakin feature?"""
4294
from bzrlib import breakin
4295
if breakin.determine_signal() is None:
4297
if sys.platform == 'win32':
4298
# Windows doesn't have os.kill, and we catch the SIGBREAK signal.
4299
# We trigger SIGBREAK via a Console api so we need ctypes to
4300
# access the function
4307
def feature_name(self):
4308
return "SIGQUIT or SIGBREAK w/ctypes on win32"
4311
BreakinFeature = _BreakinFeature()
4314
class _CaseInsCasePresFilenameFeature(Feature):
4315
"""Is the file-system case insensitive, but case-preserving?"""
4318
fileno, name = tempfile.mkstemp(prefix='MixedCase')
4320
# first check truly case-preserving for created files, then check
4321
# case insensitive when opening existing files.
4322
name = osutils.normpath(name)
4323
base, rel = osutils.split(name)
4324
found_rel = osutils.canonical_relpath(base, name)
4325
return (found_rel == rel
4326
and os.path.isfile(name.upper())
4327
and os.path.isfile(name.lower()))
4332
def feature_name(self):
4333
return "case-insensitive case-preserving filesystem"
4335
CaseInsCasePresFilenameFeature = _CaseInsCasePresFilenameFeature()
4338
class _CaseInsensitiveFilesystemFeature(Feature):
4339
"""Check if underlying filesystem is case-insensitive but *not* case
4342
# Note that on Windows, Cygwin, MacOS etc, the file-systems are far
4343
# more likely to be case preserving, so this case is rare.
4346
if CaseInsCasePresFilenameFeature.available():
4349
if TestCaseWithMemoryTransport.TEST_ROOT is None:
4350
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
4351
TestCaseWithMemoryTransport.TEST_ROOT = root
4353
root = TestCaseWithMemoryTransport.TEST_ROOT
4354
tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
4356
name_a = osutils.pathjoin(tdir, 'a')
4357
name_A = osutils.pathjoin(tdir, 'A')
4359
result = osutils.isdir(name_A)
4360
_rmtree_temp_dir(tdir)
4363
def feature_name(self):
4364
return 'case-insensitive filesystem'
4366
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
4369
# Kept for compatibility, use bzrlib.tests.features.subunit instead
4370
SubUnitFeature = _CompatabilityThunkFeature(
4371
deprecated_in((2,1,0)),
4372
'bzrlib.tests.features', 'SubUnitFeature', 'subunit')
4431
4373
# Only define SubUnitBzrRunner if subunit is available.
4433
4375
from subunit import TestProtocolClient
4434
4376
from subunit.test_results import AutoTimingTestResultDecorator
4435
class SubUnitBzrProtocolClient(TestProtocolClient):
4437
def addSuccess(self, test, details=None):
4438
# The subunit client always includes the details in the subunit
4439
# stream, but we don't want to include it in ours.
4440
if details is not None and 'log' in details:
4442
return super(SubUnitBzrProtocolClient, self).addSuccess(
4445
4377
class SubUnitBzrRunner(TextTestRunner):
4446
4378
def run(self, test):
4447
4379
result = AutoTimingTestResultDecorator(
4448
SubUnitBzrProtocolClient(self.stream))
4380
TestProtocolClient(self.stream))
4449
4381
test.run(result)
4451
4383
except ImportError:
4455
@deprecated_function(deprecated_in((2, 5, 0)))
4456
def ModuleAvailableFeature(name):
4457
from bzrlib.tests import features
4458
return features.ModuleAvailableFeature(name)
added
removed
bzrlib/tests/__init__.py
