55
50
# nb: check this before importing anything else from within it
56
51
_testtools_version = getattr(testtools, '__version__', ())
57
if _testtools_version < (0, 9, 2):
58
raise ImportError("need at least testtools 0.9.2: %s is %r"
52
if _testtools_version < (0, 9, 5):
53
raise ImportError("need at least testtools 0.9.5: %s is %r"
59
54
% (testtools.__file__, _testtools_version))
60
55
from testtools import content
62
58
from bzrlib import (
62
commands as _mod_commands,
72
plugin as _mod_plugin,
79
transport as _mod_transport,
80
import bzrlib.commands
81
import bzrlib.timestamp
83
import bzrlib.inventory
84
import bzrlib.iterablefile
87
83
import bzrlib.lsprof
88
84
except ImportError:
89
85
# lsprof not available
91
from bzrlib.merge import merge_inner
94
from bzrlib.smart import client, request, server
96
from bzrlib import symbol_versioning
87
from bzrlib.smart import client, request
88
from bzrlib.transport import (
97
92
from bzrlib.symbol_versioning import (
99
93
deprecated_function,
105
from bzrlib.transport import (
110
import bzrlib.transport
111
from bzrlib.trace import mutter, note
112
96
from bzrlib.tests import (
116
from bzrlib.tests.http_server import HttpServer
117
from bzrlib.tests.TestUtil import (
121
from bzrlib.tests.treeshape import build_tree_contents
122
101
from bzrlib.ui import NullProgressView
123
102
from bzrlib.ui.text import TextUIFactory
124
import bzrlib.version_info_formats.format_custom
125
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
127
104
# Mark this python module as being part of the implementation
128
105
# of unittest: this gives us better tracebacks where the last
140
117
SUBUNIT_SEEK_SET = 0
141
118
SUBUNIT_SEEK_CUR = 1
144
class ExtendedTestResult(unittest._TextTestResult):
120
# These are intentionally brought into this namespace. That way plugins, etc
121
# can just "from bzrlib.tests import TestCase, TestLoader, etc"
122
TestSuite = TestUtil.TestSuite
123
TestLoader = TestUtil.TestLoader
125
# Tests should run in a clean and clearly defined environment. The goal is to
126
# keep them isolated from the running environment as mush as possible. The test
127
# framework ensures the variables defined below are set (or deleted if the
128
# value is None) before a test is run and reset to their original value after
129
# the test is run. Generally if some code depends on an environment variable,
130
# the tests should start without this variable in the environment. There are a
131
# few exceptions but you shouldn't violate this rule lightly.
135
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
136
# tests do check our impls match APPDATA
137
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
141
'BZREMAIL': None, # may still be present in the environment
142
'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
143
'BZR_PROGRESS_BAR': None,
144
# This should trap leaks to ~/.bzr.log. This occurs when tests use TestCase
145
# as a base class instead of TestCaseInTempDir. Tests inheriting from
146
# TestCase should not use disk resources, BZR_LOG is one.
147
'BZR_LOG': '/you-should-use-TestCaseInTempDir-if-you-need-a-log-file',
148
'BZR_PLUGIN_PATH': None,
149
'BZR_DISABLE_PLUGINS': None,
150
'BZR_PLUGINS_AT': None,
151
'BZR_CONCURRENCY': None,
152
# Make sure that any text ui tests are consistent regardless of
153
# the environment the test case is run in; you may want tests that
154
# test other combinations. 'dumb' is a reasonable guess for tests
155
# going to a pipe or a StringIO.
161
'SSH_AUTH_SOCK': None,
171
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
172
# least. If you do (care), please update this comment
176
'BZR_REMOTE_PATH': None,
177
# Generally speaking, we don't want apport reporting on crashes in
178
# the test envirnoment unless we're specifically testing apport,
179
# so that it doesn't leak into the real system environment. We
180
# use an env var so it propagates to subprocesses.
181
'APPORT_DISABLE': '1',
185
def override_os_environ(test, env=None):
186
"""Modify os.environ keeping a copy.
188
:param test: A test instance
190
:param env: A dict containing variable definitions to be installed
193
env = isolated_environ
194
test._original_os_environ = dict([(var, value)
195
for var, value in os.environ.iteritems()])
196
for var, value in env.iteritems():
197
osutils.set_or_unset_env(var, value)
198
if var not in test._original_os_environ:
199
# The var is new, add it with a value of None, so
200
# restore_os_environ will delete it
201
test._original_os_environ[var] = None
204
def restore_os_environ(test):
205
"""Restore os.environ to its original state.
207
:param test: A test instance previously passed to override_os_environ.
209
for var, value in test._original_os_environ.iteritems():
210
# Restore the original value (or delete it if the value has been set to
211
# None in override_os_environ).
212
osutils.set_or_unset_env(var, value)
215
class ExtendedTestResult(testtools.TextTestResult):
145
216
"""Accepts, reports and accumulates the results of running tests.
147
218
Compared to the unittest version this class adds support for
196
267
self._overall_start_time = time.time()
197
268
self._strict = strict
269
self._first_thread_leaker_id = None
270
self._tests_leaking_threads_count = 0
271
self._traceback_from_test = None
199
273
def stopTestRun(self):
200
274
run = self.testsRun
201
275
actionTaken = "Ran"
202
276
stopTime = time.time()
203
277
timeTaken = stopTime - self.startTime
205
self.stream.writeln(self.separator2)
206
self.stream.writeln("%s %d test%s in %.3fs" % (actionTaken,
278
# GZ 2010-07-19: Seems testtools has no printErrors method, and though
279
# the parent class method is similar have to duplicate
280
self._show_list('ERROR', self.errors)
281
self._show_list('FAIL', self.failures)
282
self.stream.write(self.sep2)
283
self.stream.write("%s %d test%s in %.3fs\n\n" % (actionTaken,
207
284
run, run != 1 and "s" or "", timeTaken))
208
self.stream.writeln()
209
285
if not self.wasSuccessful():
210
286
self.stream.write("FAILED (")
211
287
failed, errored = map(len, (self.failures, self.errors))
218
294
if failed or errored: self.stream.write(", ")
219
295
self.stream.write("known_failure_count=%d" %
220
296
self.known_failure_count)
221
self.stream.writeln(")")
297
self.stream.write(")\n")
223
299
if self.known_failure_count:
224
self.stream.writeln("OK (known_failures=%d)" %
300
self.stream.write("OK (known_failures=%d)\n" %
225
301
self.known_failure_count)
227
self.stream.writeln("OK")
303
self.stream.write("OK\n")
228
304
if self.skip_count > 0:
229
305
skipped = self.skip_count
230
self.stream.writeln('%d test%s skipped' %
306
self.stream.write('%d test%s skipped\n' %
231
307
(skipped, skipped != 1 and "s" or ""))
232
308
if self.unsupported:
233
309
for feature, count in sorted(self.unsupported.items()):
234
self.stream.writeln("Missing feature '%s' skipped %d tests." %
310
self.stream.write("Missing feature '%s' skipped %d tests.\n" %
235
311
(feature, count))
237
313
ok = self.wasStrictlySuccessful()
239
315
ok = self.wasSuccessful()
240
if TestCase._first_thread_leaker_id:
316
if self._first_thread_leaker_id:
241
317
self.stream.write(
242
318
'%s is leaking threads among %d leaking tests.\n' % (
243
TestCase._first_thread_leaker_id,
244
TestCase._leaking_threads_tests))
319
self._first_thread_leaker_id,
320
self._tests_leaking_threads_count))
245
321
# We don't report the main thread as an active one.
246
322
self.stream.write(
247
323
'%d non-main threads were left active in the end.\n'
248
% (TestCase._active_threads - 1))
324
% (len(self._active_threads) - 1))
250
326
def getDescription(self, test):
276
353
def _shortened_test_description(self, test):
278
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
355
what = re.sub(r'^bzrlib\.tests\.', '', what)
358
# GZ 2010-10-04: Cloned tests may end up harmlessly calling this method
359
# multiple times in a row, because the handler is added for
360
# each test but the container list is shared between cases.
361
# See lp:498869 lp:625574 and lp:637725 for background.
362
def _record_traceback_from_test(self, exc_info):
363
"""Store the traceback from passed exc_info tuple till"""
364
self._traceback_from_test = exc_info[2]
281
366
def startTest(self, test):
282
unittest.TestResult.startTest(self, test)
367
super(ExtendedTestResult, self).startTest(test)
283
368
if self.count == 0:
284
369
self.startTests()
285
371
self.report_test_start(test)
286
372
test.number = self.count
287
373
self._recordTestStartTime()
374
# Make testtools cases give us the real traceback on failure
375
addOnException = getattr(test, "addOnException", None)
376
if addOnException is not None:
377
addOnException(self._record_traceback_from_test)
378
# Only check for thread leaks on bzrlib derived test cases
379
if isinstance(test, TestCase):
380
test.addCleanup(self._check_leaked_threads, test)
382
def stopTest(self, test):
383
super(ExtendedTestResult, self).stopTest(test)
384
# Manually break cycles, means touching various private things but hey
385
getDetails = getattr(test, "getDetails", None)
386
if getDetails is not None:
388
# Clear _type_equality_funcs to try to stop TestCase instances
389
# from wasting memory. 'clear' is not available in all Python
390
# versions (bug 809048)
391
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
392
if type_equality_funcs is not None:
393
tef_clear = getattr(type_equality_funcs, "clear", None)
394
if tef_clear is None:
395
tef_instance_dict = getattr(type_equality_funcs, "__dict__", None)
396
if tef_instance_dict is not None:
397
tef_clear = tef_instance_dict.clear
398
if tef_clear is not None:
400
self._traceback_from_test = None
289
402
def startTests(self):
291
if getattr(sys, 'frozen', None) is None:
292
bzr_path = osutils.realpath(sys.argv[0])
294
bzr_path = sys.executable
296
'bzr selftest: %s\n' % (bzr_path,))
299
bzrlib.__path__[0],))
301
' bzr-%s python-%s %s\n' % (
302
bzrlib.version_string,
303
bzrlib._format_version_tuple(sys.version_info),
304
platform.platform(aliased=1),
306
self.stream.write('\n')
403
self.report_tests_starting()
404
self._active_threads = threading.enumerate()
406
def _check_leaked_threads(self, test):
407
"""See if any threads have leaked since last call
409
A sample of live threads is stored in the _active_threads attribute,
410
when this method runs it compares the current live threads and any not
411
in the previous sample are treated as having leaked.
413
now_active_threads = set(threading.enumerate())
414
threads_leaked = now_active_threads.difference(self._active_threads)
416
self._report_thread_leak(test, threads_leaked, now_active_threads)
417
self._tests_leaking_threads_count += 1
418
if self._first_thread_leaker_id is None:
419
self._first_thread_leaker_id = test.id()
420
self._active_threads = now_active_threads
308
422
def _recordTestStartTime(self):
309
423
"""Record that a test has started."""
310
self._start_time = time.time()
312
def _cleanupLogFile(self, test):
313
# We can only do this if we have one of our TestCases, not if
315
setKeepLogfile = getattr(test, 'setKeepLogfile', None)
316
if setKeepLogfile is not None:
424
self._start_datetime = self._now()
319
426
def addError(self, test, err):
320
427
"""Tell result that test finished with an error.
356
461
self._formatTime(benchmark_time),
358
463
self.report_success(test)
359
self._cleanupLogFile(test)
360
unittest.TestResult.addSuccess(self, test)
464
super(ExtendedTestResult, self).addSuccess(test)
361
465
test._log_contents = ''
363
467
def addExpectedFailure(self, test, err):
364
468
self.known_failure_count += 1
365
469
self.report_known_failure(test, err)
471
def addUnexpectedSuccess(self, test, details=None):
472
"""Tell result the test unexpectedly passed, counting as a failure
474
When the minimum version of testtools required becomes 0.9.8 this
475
can be updated to use the new handling there.
477
super(ExtendedTestResult, self).addFailure(test, details=details)
478
self.failure_count += 1
479
self.report_unexpected_success(test,
480
"".join(details["reason"].iter_text()))
367
484
def addNotSupported(self, test, feature):
368
485
"""The test will not be run because of a missing feature.
401
519
raise errors.BzrError("Unknown whence %r" % whence)
403
def report_cleaning_up(self):
521
def report_tests_starting(self):
522
"""Display information before the test run begins"""
523
if getattr(sys, 'frozen', None) is None:
524
bzr_path = osutils.realpath(sys.argv[0])
526
bzr_path = sys.executable
528
'bzr selftest: %s\n' % (bzr_path,))
531
bzrlib.__path__[0],))
533
' bzr-%s python-%s %s\n' % (
534
bzrlib.version_string,
535
bzrlib._format_version_tuple(sys.version_info),
536
platform.platform(aliased=1),
538
self.stream.write('\n')
540
def report_test_start(self, test):
541
"""Display information on the test just about to be run"""
543
def _report_thread_leak(self, test, leaked_threads, active_threads):
544
"""Display information on a test that leaked one or more threads"""
545
# GZ 2010-09-09: A leak summary reported separately from the general
546
# thread debugging would be nice. Tests under subunit
547
# need something not using stream, perhaps adding a
548
# testtools details object would be fitting.
549
if 'threads' in selftest_debug_flags:
550
self.stream.write('%s is leaking, active is now %d\n' %
551
(test.id(), len(active_threads)))
406
553
def startTestRun(self):
407
554
self.startTime = time.time()
551
695
return '%s%s' % (indent, err[1])
553
697
def report_error(self, test, err):
554
self.stream.writeln('ERROR %s\n%s'
698
self.stream.write('ERROR %s\n%s\n'
555
699
% (self._testTimeString(test),
556
700
self._error_summary(err)))
558
702
def report_failure(self, test, err):
559
self.stream.writeln(' FAIL %s\n%s'
703
self.stream.write(' FAIL %s\n%s\n'
560
704
% (self._testTimeString(test),
561
705
self._error_summary(err)))
563
707
def report_known_failure(self, test, err):
564
self.stream.writeln('XFAIL %s\n%s'
708
self.stream.write('XFAIL %s\n%s\n'
565
709
% (self._testTimeString(test),
566
710
self._error_summary(err)))
712
def report_unexpected_success(self, test, reason):
713
self.stream.write(' FAIL %s\n%s: %s\n'
714
% (self._testTimeString(test),
715
"Unexpected success. Should have failed",
568
718
def report_success(self, test):
569
self.stream.writeln(' OK %s' % self._testTimeString(test))
719
self.stream.write(' OK %s\n' % self._testTimeString(test))
570
720
for bench_called, stats in getattr(test, '_benchcalls', []):
571
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
721
self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
572
722
stats.pprint(file=self.stream)
573
723
# flush the stream so that we get smooth output. This verbose mode is
574
724
# used to show the output in PQM.
575
725
self.stream.flush()
577
727
def report_skip(self, test, reason):
578
self.stream.writeln(' SKIP %s\n%s'
728
self.stream.write(' SKIP %s\n%s\n'
579
729
% (self._testTimeString(test), reason))
581
731
def report_not_applicable(self, test, reason):
582
self.stream.writeln(' N/A %s\n %s'
732
self.stream.write(' N/A %s\n %s\n'
583
733
% (self._testTimeString(test), reason))
585
735
def report_unsupported(self, test, feature):
586
736
"""test cannot be run because feature is missing."""
587
self.stream.writeln("NODEP %s\n The feature '%s' is not available."
737
self.stream.write("NODEP %s\n The feature '%s' is not available.\n"
588
738
%(self._testTimeString(test), feature))
789
961
routine, and to build and check bzr trees.
791
963
In addition to the usual method of overriding tearDown(), this class also
792
allows subclasses to register functions into the _cleanups list, which is
964
allows subclasses to register cleanup functions via addCleanup, which are
793
965
run in order as the object is torn down. It's less likely this will be
794
966
accidentally overlooked.
797
_active_threads = None
798
_leaking_threads_tests = 0
799
_first_thread_leaker_id = None
800
_log_file_name = None
801
970
# record lsprof data when performing benchmark calls.
802
971
_gather_lsprof_in_benchmarks = False
804
973
def __init__(self, methodName='testMethod'):
805
974
super(TestCase, self).__init__(methodName)
807
975
self._directory_isolation = True
808
976
self.exception_handlers.insert(0,
809
977
(UnavailableFeature, self._do_unsupported_or_skip))
827
993
self._track_transports()
828
994
self._track_locks()
829
995
self._clear_debug_flags()
830
TestCase._active_threads = threading.activeCount()
831
self.addCleanup(self._check_leaked_threads)
996
# Isolate global verbosity level, to make sure it's reproducible
997
# between tests. We should get rid of this altogether: bug 656694. --
999
self.overrideAttr(bzrlib.trace, '_verbosity_level', 0)
1000
# Isolate config option expansion until its default value for bzrlib is
1001
# settled on or a the FIXME associated with _get_expand_default_value
1002
# is addressed -- vila 20110219
1003
self.overrideAttr(config, '_expand_default_value', None)
1004
self._log_files = set()
1005
# Each key in the ``_counters`` dict holds a value for a different
1006
# counter. When the test ends, addDetail() should be used to output the
1007
# counter values. This happens in install_counter_hook().
1009
if 'config_stats' in selftest_debug_flags:
1010
self._install_config_stats_hooks()
1011
# Do not use i18n for tests (unless the test reverses this)
833
1014
def debug(self):
834
1015
# debug a frame up.
836
pdb.Pdb().set_trace(sys._getframe().f_back)
838
def _check_leaked_threads(self):
839
active = threading.activeCount()
840
leaked_threads = active - TestCase._active_threads
841
TestCase._active_threads = active
842
# If some tests make the number of threads *decrease*, we'll consider
843
# that they are just observing old threads dieing, not agressively kill
844
# random threads. So we don't report these tests as leaking. The risk
845
# is that we have false positives that way (the test see 2 threads
846
# going away but leak one) but it seems less likely than the actual
847
# false positives (the test see threads going away and does not leak).
848
if leaked_threads > 0:
849
TestCase._leaking_threads_tests += 1
850
if TestCase._first_thread_leaker_id is None:
851
TestCase._first_thread_leaker_id = self.id()
1017
# The sys preserved stdin/stdout should allow blackbox tests debugging
1018
pdb.Pdb(stdin=sys.__stdin__, stdout=sys.__stdout__
1019
).set_trace(sys._getframe().f_back)
1021
def discardDetail(self, name):
1022
"""Extend the addDetail, getDetails api so we can remove a detail.
1024
eg. bzr always adds the 'log' detail at startup, but we don't want to
1025
include it for skipped, xfail, etc tests.
1027
It is safe to call this for a detail that doesn't exist, in case this
1028
gets called multiple times.
1030
# We cheat. details is stored in __details which means we shouldn't
1031
# touch it. but getDetails() returns the dict directly, so we can
1033
details = self.getDetails()
1037
def install_counter_hook(self, hooks, name, counter_name=None):
1038
"""Install a counting hook.
1040
Any hook can be counted as long as it doesn't need to return a value.
1042
:param hooks: Where the hook should be installed.
1044
:param name: The hook name that will be counted.
1046
:param counter_name: The counter identifier in ``_counters``, defaults
1049
_counters = self._counters # Avoid closing over self
1050
if counter_name is None:
1052
if _counters.has_key(counter_name):
1053
raise AssertionError('%s is already used as a counter name'
1055
_counters[counter_name] = 0
1056
self.addDetail(counter_name, content.Content(content.UTF8_TEXT,
1057
lambda: ['%d' % (_counters[counter_name],)]))
1058
def increment_counter(*args, **kwargs):
1059
_counters[counter_name] += 1
1060
label = 'count %s calls' % (counter_name,)
1061
hooks.install_named_hook(name, increment_counter, label)
1062
self.addCleanup(hooks.uninstall_named_hook, name, label)
1064
def _install_config_stats_hooks(self):
1065
"""Install config hooks to count hook calls.
1068
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1069
self.install_counter_hook(config.ConfigHooks, hook_name,
1070
'config.%s' % (hook_name,))
1072
# The OldConfigHooks are private and need special handling to protect
1073
# against recursive tests (tests that run other tests), so we just do
1074
# manually what registering them into _builtin_known_hooks will provide
1076
self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1077
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1078
self.install_counter_hook(config.OldConfigHooks, hook_name,
1079
'old_config.%s' % (hook_name,))
853
1081
def _clear_debug_flags(self):
854
1082
"""Prevent externally set debug flags affecting tests.
866
1094
def _clear_hooks(self):
867
1095
# prevent hooks affecting tests
1096
known_hooks = hooks.known_hooks
868
1097
self._preserved_hooks = {}
869
for key, factory in hooks.known_hooks.items():
870
parent, name = hooks.known_hooks_key_to_parent_and_attribute(key)
871
current_hooks = hooks.known_hooks_key_to_object(key)
1098
for key, (parent, name) in known_hooks.iter_parent_objects():
1099
current_hooks = getattr(parent, name)
872
1100
self._preserved_hooks[parent] = (name, current_hooks)
1101
self._preserved_lazy_hooks = hooks._lazy_hooks
1102
hooks._lazy_hooks = {}
873
1103
self.addCleanup(self._restoreHooks)
874
for key, factory in hooks.known_hooks.items():
875
parent, name = hooks.known_hooks_key_to_parent_and_attribute(key)
1104
for key, (parent, name) in known_hooks.iter_parent_objects():
1105
factory = known_hooks.get(key)
876
1106
setattr(parent, name, factory())
877
1107
# this hook should always be installed
878
1108
request._install_hook()
1135
1369
'st_mtime did not match')
1136
1370
self.assertEqual(expected.st_ctime, actual.st_ctime,
1137
1371
'st_ctime did not match')
1138
if sys.platform != 'win32':
1372
if sys.platform == 'win32':
1139
1373
# On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
1140
1374
# is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
1141
# odd. Regardless we shouldn't actually try to assert anything
1142
# about their values
1375
# odd. We just force it to always be 0 to avoid any problems.
1376
self.assertEqual(0, expected.st_dev)
1377
self.assertEqual(0, actual.st_dev)
1378
self.assertEqual(0, expected.st_ino)
1379
self.assertEqual(0, actual.st_ino)
1143
1381
self.assertEqual(expected.st_dev, actual.st_dev,
1144
1382
'st_dev did not match')
1145
1383
self.assertEqual(expected.st_ino, actual.st_ino,
1322
1563
self.assertEqual(expected_docstring, obj.__doc__)
1565
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1324
1566
def failUnlessExists(self, path):
1567
return self.assertPathExists(path)
1569
def assertPathExists(self, path):
1325
1570
"""Fail unless path or paths, which may be abs or relative, exist."""
1326
1571
if not isinstance(path, basestring):
1328
self.failUnlessExists(p)
1573
self.assertPathExists(p)
1330
self.failUnless(osutils.lexists(path),path+" does not exist")
1575
self.assertTrue(osutils.lexists(path),
1576
path + " does not exist")
1578
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1332
1579
def failIfExists(self, path):
1580
return self.assertPathDoesNotExist(path)
1582
def assertPathDoesNotExist(self, path):
1333
1583
"""Fail if path or paths, which may be abs or relative, exist."""
1334
1584
if not isinstance(path, basestring):
1336
self.failIfExists(p)
1586
self.assertPathDoesNotExist(p)
1338
self.failIf(osutils.lexists(path),path+" exists")
1588
self.assertFalse(osutils.lexists(path),
1340
1591
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1341
1592
"""A helper for callDeprecated and applyDeprecated.
1456
1708
The file is removed as the test is torn down.
1458
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
1459
self._log_file = os.fdopen(fileno, 'w+')
1460
self._log_memento = bzrlib.trace.push_log_file(self._log_file)
1461
self._log_file_name = name
1710
pseudo_log_file = StringIO()
1711
def _get_log_contents_for_weird_testtools_api():
1712
return [pseudo_log_file.getvalue().decode(
1713
"utf-8", "replace").encode("utf-8")]
1714
self.addDetail("log", content.Content(content.ContentType("text",
1715
"plain", {"charset": "utf8"}),
1716
_get_log_contents_for_weird_testtools_api))
1717
self._log_file = pseudo_log_file
1718
self._log_memento = trace.push_log_file(self._log_file)
1462
1719
self.addCleanup(self._finishLogFile)
1464
1721
def _finishLogFile(self):
1465
1722
"""Finished with the log file.
1467
Close the file and delete it, unless setKeepLogfile was called.
1724
Close the file and delete it.
1469
if bzrlib.trace._trace_file:
1726
if trace._trace_file:
1470
1727
# flush the log file, to get all content
1471
bzrlib.trace._trace_file.flush()
1472
bzrlib.trace.pop_log_file(self._log_memento)
1473
# Cache the log result and delete the file on disk
1474
self._get_log(False)
1728
trace._trace_file.flush()
1729
trace.pop_log_file(self._log_memento)
1476
1731
def thisFailsStrictLockCheck(self):
1477
1732
"""It is known that this test would fail with -Dstrict_locks.
1513
1763
setattr(obj, attr_name, new)
1766
def overrideEnv(self, name, new):
1767
"""Set an environment variable, and reset it after the test.
1769
:param name: The environment variable name.
1771
:param new: The value to set the variable to. If None, the
1772
variable is deleted from the environment.
1774
:returns: The actual variable value.
1776
value = osutils.set_or_unset_env(name, new)
1777
self.addCleanup(osutils.set_or_unset_env, name, value)
1780
def recordCalls(self, obj, attr_name):
1781
"""Monkeypatch in a wrapper that will record calls.
1783
The monkeypatch is automatically removed when the test concludes.
1785
:param obj: The namespace holding the reference to be replaced;
1786
typically a module, class, or object.
1787
:param attr_name: A string for the name of the attribute to
1789
:returns: A list that will be extended with one item every time the
1790
function is called, with a tuple of (args, kwargs).
1794
def decorator(*args, **kwargs):
1795
calls.append((args, kwargs))
1796
return orig(*args, **kwargs)
1797
orig = self.overrideAttr(obj, attr_name, decorator)
1516
1800
def _cleanEnvironment(self):
1518
'BZR_HOME': None, # Don't inherit BZR_HOME to all the tests.
1519
'HOME': os.getcwd(),
1520
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
1521
# tests do check our impls match APPDATA
1522
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
1526
'BZREMAIL': None, # may still be present in the environment
1528
'BZR_PROGRESS_BAR': None,
1530
'BZR_PLUGIN_PATH': None,
1531
'BZR_DISABLE_PLUGINS': None,
1532
'BZR_PLUGINS_AT': None,
1533
'BZR_CONCURRENCY': None,
1534
# Make sure that any text ui tests are consistent regardless of
1535
# the environment the test case is run in; you may want tests that
1536
# test other combinations. 'dumb' is a reasonable guess for tests
1537
# going to a pipe or a StringIO.
1541
'BZR_COLUMNS': '80',
1543
'SSH_AUTH_SOCK': None,
1547
'https_proxy': None,
1548
'HTTPS_PROXY': None,
1553
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
1554
# least. If you do (care), please update this comment
1558
'BZR_REMOTE_PATH': None,
1559
# Generally speaking, we don't want apport reporting on crashes in
1560
# the test envirnoment unless we're specifically testing apport,
1561
# so that it doesn't leak into the real system environment. We
1562
# use an env var so it propagates to subprocesses.
1563
'APPORT_DISABLE': '1',
1566
self.addCleanup(self._restoreEnvironment)
1567
for name, value in new_env.iteritems():
1568
self._captureVar(name, value)
1570
def _captureVar(self, name, newvalue):
1571
"""Set an environment variable, and reset it when finished."""
1572
self._old_env[name] = osutils.set_or_unset_env(name, newvalue)
1574
def _restoreEnvironment(self):
1575
for name, value in self._old_env.iteritems():
1576
osutils.set_or_unset_env(name, value)
1801
for name, value in isolated_environ.iteritems():
1802
self.overrideEnv(name, value)
1578
1804
def _restoreHooks(self):
1579
1805
for klass, (name, hooks) in self._preserved_hooks.items():
1580
1806
setattr(klass, name, hooks)
1807
self._preserved_hooks.clear()
1808
bzrlib.hooks._lazy_hooks = self._preserved_lazy_hooks
1809
self._preserved_lazy_hooks.clear()
1582
1811
def knownFailure(self, reason):
1583
"""This test has failed for some known reason."""
1584
raise KnownFailure(reason)
1812
"""Declare that this test fails for a known reason
1814
Tests that are known to fail should generally be using expectedFailure
1815
with an appropriate reverse assertion if a change could cause the test
1816
to start passing. Conversely if the test has no immediate prospect of
1817
succeeding then using skip is more suitable.
1819
When this method is called while an exception is being handled, that
1820
traceback will be used, otherwise a new exception will be thrown to
1821
provide one but won't be reported.
1823
self._add_reason(reason)
1825
exc_info = sys.exc_info()
1826
if exc_info != (None, None, None):
1827
self._report_traceback(exc_info)
1830
raise self.failureException(reason)
1831
except self.failureException:
1832
exc_info = sys.exc_info()
1833
# GZ 02-08-2011: Maybe cleanup this err.exc_info attribute too?
1834
raise testtools.testcase._ExpectedFailure(exc_info)
1838
def _suppress_log(self):
1839
"""Remove the log info from details."""
1840
self.discardDetail('log')
1586
1842
def _do_skip(self, result, reason):
1843
self._suppress_log()
1587
1844
addSkip = getattr(result, 'addSkip', None)
1588
1845
if not callable(addSkip):
1589
1846
result.addSuccess(result)
1612
1871
self._do_skip(result, reason)
1874
def _report_skip(self, result, err):
1875
"""Override the default _report_skip.
1877
We want to strip the 'log' detail. If we waint until _do_skip, it has
1878
already been formatted into the 'reason' string, and we can't pull it
1881
self._suppress_log()
1882
super(TestCase, self)._report_skip(self, result, err)
1885
def _report_expected_failure(self, result, err):
1888
See _report_skip for motivation.
1890
self._suppress_log()
1891
super(TestCase, self)._report_expected_failure(self, result, err)
1615
1894
def _do_unsupported_or_skip(self, result, e):
1616
1895
reason = e.args[0]
1896
self._suppress_log()
1617
1897
addNotSupported = getattr(result, 'addNotSupported', None)
1618
1898
if addNotSupported is not None:
1619
1899
result.addNotSupported(self, reason)
1645
1925
self._benchtime += time.time() - start
1647
1927
def log(self, *args):
1650
def _get_log(self, keep_log_file=False):
1651
"""Internal helper to get the log from bzrlib.trace for this test.
1653
Please use self.getDetails, or self.get_log to access this in test case
1656
:param keep_log_file: When True, if the log is still a file on disk
1657
leave it as a file on disk. When False, if the log is still a file
1658
on disk, the log file is deleted and the log preserved as
1660
:return: A string containing the log.
1662
if self._log_contents is not None:
1664
self._log_contents.decode('utf8')
1665
except UnicodeDecodeError:
1666
unicodestr = self._log_contents.decode('utf8', 'replace')
1667
self._log_contents = unicodestr.encode('utf8')
1668
return self._log_contents
1670
if bzrlib.trace._trace_file:
1671
# flush the log file, to get all content
1672
bzrlib.trace._trace_file.flush()
1673
if self._log_file_name is not None:
1674
logfile = open(self._log_file_name)
1676
log_contents = logfile.read()
1680
log_contents.decode('utf8')
1681
except UnicodeDecodeError:
1682
unicodestr = log_contents.decode('utf8', 'replace')
1683
log_contents = unicodestr.encode('utf8')
1684
if not keep_log_file:
1686
max_close_attempts = 100
1687
first_close_error = None
1688
while close_attempts < max_close_attempts:
1691
self._log_file.close()
1692
except IOError, ioe:
1693
if ioe.errno is None:
1694
# No errno implies 'close() called during
1695
# concurrent operation on the same file object', so
1696
# retry. Probably a thread is trying to write to
1698
if first_close_error is None:
1699
first_close_error = ioe
1704
if close_attempts > 1:
1706
'Unable to close log file on first attempt, '
1707
'will retry: %s\n' % (first_close_error,))
1708
if close_attempts == max_close_attempts:
1710
'Unable to close log file after %d attempts.\n'
1711
% (max_close_attempts,))
1712
self._log_file = None
1713
# Permit multiple calls to get_log until we clean it up in
1715
self._log_contents = log_contents
1717
os.remove(self._log_file_name)
1719
if sys.platform == 'win32' and e.errno == errno.EACCES:
1720
sys.stderr.write(('Unable to delete log file '
1721
' %r\n' % self._log_file_name))
1724
self._log_file_name = None
1727
return "No log file content and no log file name."
1729
1930
def get_log(self):
1730
1931
"""Get a unicode string containing the log from bzrlib.trace.
2207
def _add_subprocess_log(self, log_file_path):
2208
if len(self._log_files) == 0:
2209
# Register an addCleanup func. We do this on the first call to
2210
# _add_subprocess_log rather than in TestCase.setUp so that this
2211
# addCleanup is registered after any cleanups for tempdirs that
2212
# subclasses might create, which will probably remove the log file
2214
self.addCleanup(self._subprocess_log_cleanup)
2215
# self._log_files is a set, so if a log file is reused we won't grab it
2217
self._log_files.add(log_file_path)
2219
def _subprocess_log_cleanup(self):
2220
for count, log_file_path in enumerate(self._log_files):
2221
# We use buffer_now=True to avoid holding the file open beyond
2222
# the life of this function, which might interfere with e.g.
2223
# cleaning tempdirs on Windows.
2224
# XXX: Testtools 0.9.5 doesn't have the content_from_file helper
2225
#detail_content = content.content_from_file(
2226
# log_file_path, buffer_now=True)
2227
with open(log_file_path, 'rb') as log_file:
2228
log_file_bytes = log_file.read()
2229
detail_content = content.Content(content.ContentType("text",
2230
"plain", {"charset": "utf8"}), lambda: [log_file_bytes])
2231
self.addDetail("start_bzr_subprocess-log-%d" % (count,),
1997
2234
def _popen(self, *args, **kwargs):
1998
2235
"""Place a call to Popen.
2000
2237
Allows tests to override this method to intercept the calls made to
2001
2238
Popen for introspection.
2003
return Popen(*args, **kwargs)
2240
return subprocess.Popen(*args, **kwargs)
2005
2242
def get_source_path(self):
2006
2243
"""Return the path of the directory containing bzrlib."""
2036
2273
if retcode is not None and retcode != process.returncode:
2037
2274
if process_args is None:
2038
2275
process_args = "(unknown args)"
2039
mutter('Output of bzr %s:\n%s', process_args, out)
2040
mutter('Error for bzr %s:\n%s', process_args, err)
2276
trace.mutter('Output of bzr %s:\n%s', process_args, out)
2277
trace.mutter('Error for bzr %s:\n%s', process_args, err)
2041
2278
self.fail('Command bzr %s failed with retcode %s != %s'
2042
2279
% (process_args, retcode, process.returncode))
2043
2280
return [out, err]
2045
def check_inventory_shape(self, inv, shape):
2046
"""Compare an inventory to a list of expected names.
2282
def check_tree_shape(self, tree, shape):
2283
"""Compare a tree to a list of expected names.
2048
2285
Fail if they are not precisely equal.
2051
2288
shape = list(shape) # copy
2052
for path, ie in inv.entries():
2289
for path, ie in tree.iter_entries_by_dir():
2053
2290
name = path.replace('\\', '/')
2054
2291
if ie.kind == 'directory':
2055
2292
name = name + '/'
2294
pass # ignore root entry
2057
2296
shape.remove(name)
2059
2298
extras.append(name)
2149
2388
class TestCaseWithMemoryTransport(TestCase):
2150
2389
"""Common test class for tests that do not need disk resources.
2152
Tests that need disk resources should derive from TestCaseWithTransport.
2391
Tests that need disk resources should derive from TestCaseInTempDir
2392
orTestCaseWithTransport.
2154
2394
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
2156
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
2396
For TestCaseWithMemoryTransport the ``test_home_dir`` is set to the name of
2157
2397
a directory which does not exist. This serves to help ensure test isolation
2158
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
2159
must exist. However, TestCaseWithMemoryTransport does not offer local
2160
file defaults for the transport in tests, nor does it obey the command line
2398
is preserved. ``test_dir`` is set to the TEST_ROOT, as is cwd, because they
2399
must exist. However, TestCaseWithMemoryTransport does not offer local file
2400
defaults for the transport in tests, nor does it obey the command line
2161
2401
override, so tests that accidentally write to the common directory should
2164
:cvar TEST_ROOT: Directory containing all temporary directories, plus
2165
a .bzr directory that stops us ascending higher into the filesystem.
2404
:cvar TEST_ROOT: Directory containing all temporary directories, plus a
2405
``.bzr`` directory that stops us ascending higher into the filesystem.
2168
2408
TEST_ROOT = None
2378
2632
def make_branch(self, relpath, format=None):
2379
2633
"""Create a branch on the transport at relpath."""
2380
2634
repo = self.make_repository(relpath, format=format)
2381
return repo.bzrdir.create_branch()
2635
return repo.bzrdir.create_branch(append_revisions_only=False)
2637
def resolve_format(self, format):
2638
"""Resolve an object to a ControlDir format object.
2640
The initial format object can either already be
2641
a ControlDirFormat, None (for the default format),
2642
or a string with the name of the control dir format.
2644
:param format: Object to resolve
2645
:return A ControlDirFormat instance
2649
if isinstance(format, basestring):
2650
format = bzrdir.format_registry.make_bzrdir(format)
2653
def resolve_format(self, format):
2654
"""Resolve an object to a ControlDir format object.
2656
The initial format object can either already be
2657
a ControlDirFormat, None (for the default format),
2658
or a string with the name of the control dir format.
2660
:param format: Object to resolve
2661
:return A ControlDirFormat instance
2665
if isinstance(format, basestring):
2666
format = bzrdir.format_registry.make_bzrdir(format)
2383
2669
def make_bzrdir(self, relpath, format=None):
2385
2671
# might be a relative or absolute path
2386
2672
maybe_a_url = self.get_url(relpath)
2387
2673
segments = maybe_a_url.rsplit('/', 1)
2388
t = get_transport(maybe_a_url)
2674
t = _mod_transport.get_transport(maybe_a_url)
2389
2675
if len(segments) > 1 and segments[-1] not in ('', '.'):
2390
2676
t.ensure_base()
2393
if isinstance(format, basestring):
2394
format = bzrdir.format_registry.make_bzrdir(format)
2677
format = self.resolve_format(format)
2395
2678
return format.initialize_on_transport(t)
2396
2679
except errors.UninitializableFormat:
2397
2680
raise TestSkipped("Format %s is not initializable." % format)
2399
def make_repository(self, relpath, shared=False, format=None):
2682
def make_repository(self, relpath, shared=None, format=None):
2400
2683
"""Create a repository on our default transport at relpath.
2402
2685
Note that relpath must be a relative path, not a full url.
2427
2713
test_home_dir = self.test_home_dir
2428
2714
if isinstance(test_home_dir, unicode):
2429
2715
test_home_dir = test_home_dir.encode(sys.getfilesystemencoding())
2430
os.environ['HOME'] = test_home_dir
2431
os.environ['BZR_HOME'] = test_home_dir
2716
self.overrideEnv('HOME', test_home_dir)
2717
self.overrideEnv('BZR_HOME', test_home_dir)
2433
2719
def setUp(self):
2434
2720
super(TestCaseWithMemoryTransport, self).setUp()
2721
# Ensure that ConnectedTransport doesn't leak sockets
2722
def get_transport_from_url_with_cleanup(*args, **kwargs):
2723
t = orig_get_transport_from_url(*args, **kwargs)
2724
if isinstance(t, _mod_transport.ConnectedTransport):
2725
self.addCleanup(t.disconnect)
2728
orig_get_transport_from_url = self.overrideAttr(
2729
_mod_transport, 'get_transport_from_url',
2730
get_transport_from_url_with_cleanup)
2435
2731
self._make_test_root()
2436
2732
self.addCleanup(os.chdir, os.getcwdu())
2437
2733
self.makeAndChdirToTestDir()
3191
3499
def partition_tests(suite, count):
3192
3500
"""Partition suite into count lists of tests."""
3194
tests = list(iter_suite_tests(suite))
3195
tests_per_process = int(math.ceil(float(len(tests)) / count))
3196
for block in range(count):
3197
low_test = block * tests_per_process
3198
high_test = low_test + tests_per_process
3199
process_tests = tests[low_test:high_test]
3200
result.append(process_tests)
3501
# This just assigns tests in a round-robin fashion. On one hand this
3502
# splits up blocks of related tests that might run faster if they shared
3503
# resources, but on the other it avoids assigning blocks of slow tests to
3504
# just one partition. So the slowest partition shouldn't be much slower
3506
partitions = [list() for i in range(count)]
3507
tests = iter_suite_tests(suite)
3508
for partition, test in itertools.izip(itertools.cycle(partitions), tests):
3509
partition.append(test)
3204
3513
def workaround_zealous_crypto_random():
3325
class ForwardingResult(unittest.TestResult):
3327
def __init__(self, target):
3328
unittest.TestResult.__init__(self)
3329
self.result = target
3331
def startTest(self, test):
3332
self.result.startTest(test)
3334
def stopTest(self, test):
3335
self.result.stopTest(test)
3337
def startTestRun(self):
3338
self.result.startTestRun()
3340
def stopTestRun(self):
3341
self.result.stopTestRun()
3343
def addSkip(self, test, reason):
3344
self.result.addSkip(test, reason)
3346
def addSuccess(self, test):
3347
self.result.addSuccess(test)
3349
def addError(self, test, err):
3350
self.result.addError(test, err)
3352
def addFailure(self, test, err):
3353
self.result.addFailure(test, err)
3354
ForwardingResult = testtools.ExtendedToOriginalDecorator
3357
class ProfileResult(ForwardingResult):
3636
class ProfileResult(testtools.ExtendedToOriginalDecorator):
3358
3637
"""Generate profiling data for all activity between start and success.
3360
3639
The profile data is appended to the test's _benchcalls attribute and can
3648
3937
'bzrlib.tests.per_repository',
3649
3938
'bzrlib.tests.per_repository_chk',
3650
3939
'bzrlib.tests.per_repository_reference',
3940
'bzrlib.tests.per_repository_vf',
3651
3941
'bzrlib.tests.per_uifactory',
3652
3942
'bzrlib.tests.per_versionedfile',
3653
3943
'bzrlib.tests.per_workingtree',
3654
3944
'bzrlib.tests.test__annotator',
3655
3945
'bzrlib.tests.test__bencode',
3946
'bzrlib.tests.test__btree_serializer',
3656
3947
'bzrlib.tests.test__chk_map',
3657
3948
'bzrlib.tests.test__dirstate_helpers',
3658
3949
'bzrlib.tests.test__groupcompress',
3686
3977
'bzrlib.tests.test_commit_merge',
3687
3978
'bzrlib.tests.test_config',
3688
3979
'bzrlib.tests.test_conflicts',
3980
'bzrlib.tests.test_controldir',
3689
3981
'bzrlib.tests.test_counted_lock',
3690
3982
'bzrlib.tests.test_crash',
3691
3983
'bzrlib.tests.test_decorators',
3692
3984
'bzrlib.tests.test_delta',
3693
3985
'bzrlib.tests.test_debug',
3694
'bzrlib.tests.test_deprecated_graph',
3695
3986
'bzrlib.tests.test_diff',
3696
3987
'bzrlib.tests.test_directory_service',
3697
3988
'bzrlib.tests.test_dirstate',
3698
3989
'bzrlib.tests.test_email_message',
3699
3990
'bzrlib.tests.test_eol_filters',
3700
3991
'bzrlib.tests.test_errors',
3992
'bzrlib.tests.test_estimate_compressed_size',
3701
3993
'bzrlib.tests.test_export',
3994
'bzrlib.tests.test_export_pot',
3702
3995
'bzrlib.tests.test_extract',
3996
'bzrlib.tests.test_features',
3703
3997
'bzrlib.tests.test_fetch',
3998
'bzrlib.tests.test_fixtures',
3704
3999
'bzrlib.tests.test_fifo_cache',
3705
4000
'bzrlib.tests.test_filters',
4001
'bzrlib.tests.test_filter_tree',
3706
4002
'bzrlib.tests.test_ftp_transport',
3707
4003
'bzrlib.tests.test_foreign',
3708
4004
'bzrlib.tests.test_generate_docs',
3938
4247
# Some tests mentioned in the list are not in the test suite. The
3939
4248
# list may be out of date, report to the tester.
3940
4249
for id in not_found:
3941
bzrlib.trace.warning('"%s" not found in the test suite', id)
4250
trace.warning('"%s" not found in the test suite', id)
3942
4251
for id in duplicates:
3943
bzrlib.trace.warning('"%s" is used as an id by several tests', id)
4252
trace.warning('"%s" is used as an id by several tests', id)
3948
def multiply_scenarios(scenarios_left, scenarios_right):
4257
def multiply_scenarios(*scenarios):
4258
"""Multiply two or more iterables of scenarios.
4260
It is safe to pass scenario generators or iterators.
4262
:returns: A list of compound scenarios: the cross-product of all
4263
scenarios, with the names concatenated and the parameters
4266
return reduce(_multiply_two_scenarios, map(list, scenarios))
4269
def _multiply_two_scenarios(scenarios_left, scenarios_right):
3949
4270
"""Multiply two sets of scenarios.
3951
4272
:returns: the cartesian product of the two sets of scenarios, that is
4102
4436
if test_id != None:
4103
4437
ui.ui_factory.clear_term()
4104
4438
sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4439
# Ugly, but the last thing we want here is fail, so bear with it.
4440
printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4441
).encode('ascii', 'replace')
4105
4442
sys.stderr.write('Unable to remove testing dir %s\n%s'
4106
% (os.path.basename(dirname), e))
4109
class Feature(object):
4110
"""An operating system Feature."""
4113
self._available = None
4115
def available(self):
4116
"""Is the feature available?
4118
:return: True if the feature is available.
4120
if self._available is None:
4121
self._available = self._probe()
4122
return self._available
4125
"""Implement this method in concrete features.
4127
:return: True if the feature is available.
4129
raise NotImplementedError
4132
if getattr(self, 'feature_name', None):
4133
return self.feature_name()
4134
return self.__class__.__name__
4137
class _SymlinkFeature(Feature):
4140
return osutils.has_symlinks()
4142
def feature_name(self):
4145
SymlinkFeature = _SymlinkFeature()
4148
class _HardlinkFeature(Feature):
4151
return osutils.has_hardlinks()
4153
def feature_name(self):
4156
HardlinkFeature = _HardlinkFeature()
4159
class _OsFifoFeature(Feature):
4162
return getattr(os, 'mkfifo', None)
4164
def feature_name(self):
4165
return 'filesystem fifos'
4167
OsFifoFeature = _OsFifoFeature()
4170
class _UnicodeFilenameFeature(Feature):
4171
"""Does the filesystem support Unicode filenames?"""
4175
# Check for character combinations unlikely to be covered by any
4176
# single non-unicode encoding. We use the characters
4177
# - greek small letter alpha (U+03B1) and
4178
# - braille pattern dots-123456 (U+283F).
4179
os.stat(u'\u03b1\u283f')
4180
except UnicodeEncodeError:
4182
except (IOError, OSError):
4183
# The filesystem allows the Unicode filename but the file doesn't
4187
# The filesystem allows the Unicode filename and the file exists,
4191
UnicodeFilenameFeature = _UnicodeFilenameFeature()
4194
class _CompatabilityThunkFeature(Feature):
4195
"""This feature is just a thunk to another feature.
4197
It issues a deprecation warning if it is accessed, to let you know that you
4198
should really use a different feature.
4201
def __init__(self, dep_version, module, name,
4202
replacement_name, replacement_module=None):
4203
super(_CompatabilityThunkFeature, self).__init__()
4204
self._module = module
4205
if replacement_module is None:
4206
replacement_module = module
4207
self._replacement_module = replacement_module
4209
self._replacement_name = replacement_name
4210
self._dep_version = dep_version
4211
self._feature = None
4214
if self._feature is None:
4215
depr_msg = self._dep_version % ('%s.%s'
4216
% (self._module, self._name))
4217
use_msg = ' Use %s.%s instead.' % (self._replacement_module,
4218
self._replacement_name)
4219
symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
4220
# Import the new feature and use it as a replacement for the
4222
mod = __import__(self._replacement_module, {}, {},
4223
[self._replacement_name])
4224
self._feature = getattr(mod, self._replacement_name)
4228
return self._feature._probe()
4231
class ModuleAvailableFeature(Feature):
4232
"""This is a feature than describes a module we want to be available.
4234
Declare the name of the module in __init__(), and then after probing, the
4235
module will be available as 'self.module'.
4237
:ivar module: The module if it is available, else None.
4240
def __init__(self, module_name):
4241
super(ModuleAvailableFeature, self).__init__()
4242
self.module_name = module_name
4246
self._module = __import__(self.module_name, {}, {}, [''])
4253
if self.available(): # Make sure the probe has been done
4257
def feature_name(self):
4258
return self.module_name
4261
# This is kept here for compatibility, it is recommended to use
4262
# 'bzrlib.tests.feature.paramiko' instead
4263
ParamikoFeature = _CompatabilityThunkFeature(
4264
deprecated_in((2,1,0)),
4265
'bzrlib.tests.features', 'ParamikoFeature', 'paramiko')
4443
% (os.path.basename(dirname), printable_e))
4268
4446
def probe_unicode_in_user_encoding():
4301
class _HTTPSServerFeature(Feature):
4302
"""Some tests want an https Server, check if one is available.
4304
Right now, the only way this is available is under python2.6 which provides
4315
def feature_name(self):
4316
return 'HTTPSServer'
4319
HTTPSServerFeature = _HTTPSServerFeature()
4322
class _UnicodeFilename(Feature):
4323
"""Does the filesystem support Unicode filenames?"""
4328
except UnicodeEncodeError:
4330
except (IOError, OSError):
4331
# The filesystem allows the Unicode filename but the file doesn't
4335
# The filesystem allows the Unicode filename and the file exists,
4339
UnicodeFilename = _UnicodeFilename()
4342
class _UTF8Filesystem(Feature):
4343
"""Is the filesystem UTF-8?"""
4346
if osutils._fs_enc.upper() in ('UTF-8', 'UTF8'):
4350
UTF8Filesystem = _UTF8Filesystem()
4353
class _BreakinFeature(Feature):
4354
"""Does this platform support the breakin feature?"""
4357
from bzrlib import breakin
4358
if breakin.determine_signal() is None:
4360
if sys.platform == 'win32':
4361
# Windows doesn't have os.kill, and we catch the SIGBREAK signal.
4362
# We trigger SIGBREAK via a Console api so we need ctypes to
4363
# access the function
4370
def feature_name(self):
4371
return "SIGQUIT or SIGBREAK w/ctypes on win32"
4374
BreakinFeature = _BreakinFeature()
4377
class _CaseInsCasePresFilenameFeature(Feature):
4378
"""Is the file-system case insensitive, but case-preserving?"""
4381
fileno, name = tempfile.mkstemp(prefix='MixedCase')
4383
# first check truly case-preserving for created files, then check
4384
# case insensitive when opening existing files.
4385
name = osutils.normpath(name)
4386
base, rel = osutils.split(name)
4387
found_rel = osutils.canonical_relpath(base, name)
4388
return (found_rel == rel
4389
and os.path.isfile(name.upper())
4390
and os.path.isfile(name.lower()))
4395
def feature_name(self):
4396
return "case-insensitive case-preserving filesystem"
4398
CaseInsCasePresFilenameFeature = _CaseInsCasePresFilenameFeature()
4401
class _CaseInsensitiveFilesystemFeature(Feature):
4402
"""Check if underlying filesystem is case-insensitive but *not* case
4405
# Note that on Windows, Cygwin, MacOS etc, the file-systems are far
4406
# more likely to be case preserving, so this case is rare.
4409
if CaseInsCasePresFilenameFeature.available():
4412
if TestCaseWithMemoryTransport.TEST_ROOT is None:
4413
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
4414
TestCaseWithMemoryTransport.TEST_ROOT = root
4416
root = TestCaseWithMemoryTransport.TEST_ROOT
4417
tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
4419
name_a = osutils.pathjoin(tdir, 'a')
4420
name_A = osutils.pathjoin(tdir, 'A')
4422
result = osutils.isdir(name_A)
4423
_rmtree_temp_dir(tdir)
4426
def feature_name(self):
4427
return 'case-insensitive filesystem'
4429
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
4432
class _CaseSensitiveFilesystemFeature(Feature):
4435
if CaseInsCasePresFilenameFeature.available():
4437
elif CaseInsensitiveFilesystemFeature.available():
4442
def feature_name(self):
4443
return 'case-sensitive filesystem'
4445
# new coding style is for feature instances to be lowercase
4446
case_sensitive_filesystem_feature = _CaseSensitiveFilesystemFeature()
4449
# Kept for compatibility, use bzrlib.tests.features.subunit instead
4450
SubUnitFeature = _CompatabilityThunkFeature(
4451
deprecated_in((2,1,0)),
4452
'bzrlib.tests.features', 'SubUnitFeature', 'subunit')
4453
4479
# Only define SubUnitBzrRunner if subunit is available.
4455
4481
from subunit import TestProtocolClient
4456
4482
from subunit.test_results import AutoTimingTestResultDecorator
4483
class SubUnitBzrProtocolClient(TestProtocolClient):
4485
def addSuccess(self, test, details=None):
4486
# The subunit client always includes the details in the subunit
4487
# stream, but we don't want to include it in ours.
4488
if details is not None and 'log' in details:
4490
return super(SubUnitBzrProtocolClient, self).addSuccess(
4457
4493
class SubUnitBzrRunner(TextTestRunner):
4458
4494
def run(self, test):
4459
4495
result = AutoTimingTestResultDecorator(
4460
TestProtocolClient(self.stream))
4496
SubUnitBzrProtocolClient(self.stream))
4461
4497
test.run(result)
4463
4499
except ImportError:
4466
class _PosixPermissionsFeature(Feature):
4470
# create temporary file and check if specified perms are maintained.
4473
write_perms = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR
4474
f = tempfile.mkstemp(prefix='bzr_perms_chk_')
4477
os.chmod(name, write_perms)
4479
read_perms = os.stat(name).st_mode & 0777
4481
return (write_perms == read_perms)
4483
return (os.name == 'posix') and has_perms()
4485
def feature_name(self):
4486
return 'POSIX permissions support'
4488
posix_permissions_feature = _PosixPermissionsFeature()
4503
@deprecated_function(deprecated_in((2, 5, 0)))
4504
def ModuleAvailableFeature(name):
4505
from bzrlib.tests import features
4506
return features.ModuleAvailableFeature(name)
added
removed
bzrlib/tests/__init__.py
