55
53
# nb: check this before importing anything else from within it
56
54
_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"
55
if _testtools_version < (0, 9, 5):
56
raise ImportError("need at least testtools 0.9.5: %s is %r"
59
57
% (testtools.__file__, _testtools_version))
60
58
from testtools import content
62
61
from bzrlib import (
65
commands as _mod_commands,
75
plugin as _mod_plugin,
82
transport as _mod_transport,
80
import bzrlib.commands
81
import bzrlib.timestamp
83
import bzrlib.inventory
84
import bzrlib.iterablefile
87
86
import bzrlib.lsprof
88
87
except ImportError:
89
88
# lsprof not available
91
from bzrlib.merge import merge_inner
94
from bzrlib.smart import client, request, server
96
from bzrlib import symbol_versioning
90
from bzrlib.smart import client, request
91
from bzrlib.transport import (
97
95
from bzrlib.symbol_versioning import (
99
96
deprecated_function,
105
from bzrlib.transport import (
110
import bzrlib.transport
111
from bzrlib.trace import mutter, note
112
99
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
105
from bzrlib.ui import NullProgressView
123
106
from bzrlib.ui.text import TextUIFactory
124
import bzrlib.version_info_formats.format_custom
125
from bzrlib.workingtree import WorkingTree, WorkingTreeFormat2
107
from bzrlib.tests.features import _CompatabilityThunkFeature
127
109
# Mark this python module as being part of the implementation
128
110
# of unittest: this gives us better tracebacks where the last
140
122
SUBUNIT_SEEK_SET = 0
141
123
SUBUNIT_SEEK_CUR = 1
144
class ExtendedTestResult(unittest._TextTestResult):
125
# These are intentionally brought into this namespace. That way plugins, etc
126
# can just "from bzrlib.tests import TestCase, TestLoader, etc"
127
TestSuite = TestUtil.TestSuite
128
TestLoader = TestUtil.TestLoader
130
# Tests should run in a clean and clearly defined environment. The goal is to
131
# keep them isolated from the running environment as mush as possible. The test
132
# framework ensures the variables defined below are set (or deleted if the
133
# value is None) before a test is run and reset to their original value after
134
# the test is run. Generally if some code depends on an environment variable,
135
# the tests should start without this variable in the environment. There are a
136
# few exceptions but you shouldn't violate this rule lightly.
140
# bzr now uses the Win32 API and doesn't rely on APPDATA, but the
141
# tests do check our impls match APPDATA
142
'BZR_EDITOR': None, # test_msgeditor manipulates this variable
146
'BZREMAIL': None, # may still be present in the environment
147
'EMAIL': 'jrandom@example.com', # set EMAIL as bzr does not guess
148
'BZR_PROGRESS_BAR': None,
149
# This should trap leaks to ~/.bzr.log. This occurs when tests use TestCase
150
# as a base class instead of TestCaseInTempDir. Tests inheriting from
151
# TestCase should not use disk resources, BZR_LOG is one.
152
'BZR_LOG': '/you-should-use-TestCaseInTempDir-if-you-need-a-log-file',
153
'BZR_PLUGIN_PATH': None,
154
'BZR_DISABLE_PLUGINS': None,
155
'BZR_PLUGINS_AT': None,
156
'BZR_CONCURRENCY': None,
157
# Make sure that any text ui tests are consistent regardless of
158
# the environment the test case is run in; you may want tests that
159
# test other combinations. 'dumb' is a reasonable guess for tests
160
# going to a pipe or a StringIO.
166
'SSH_AUTH_SOCK': None,
176
# Nobody cares about ftp_proxy, FTP_PROXY AFAIK. So far at
177
# least. If you do (care), please update this comment
181
'BZR_REMOTE_PATH': None,
182
# Generally speaking, we don't want apport reporting on crashes in
183
# the test envirnoment unless we're specifically testing apport,
184
# so that it doesn't leak into the real system environment. We
185
# use an env var so it propagates to subprocesses.
186
'APPORT_DISABLE': '1',
190
def override_os_environ(test, env=None):
191
"""Modify os.environ keeping a copy.
193
:param test: A test instance
195
:param env: A dict containing variable definitions to be installed
198
env = isolated_environ
199
test._original_os_environ = dict([(var, value)
200
for var, value in os.environ.iteritems()])
201
for var, value in env.iteritems():
202
osutils.set_or_unset_env(var, value)
203
if var not in test._original_os_environ:
204
# The var is new, add it with a value of None, so
205
# restore_os_environ will delete it
206
test._original_os_environ[var] = None
209
def restore_os_environ(test):
210
"""Restore os.environ to its original state.
212
:param test: A test instance previously passed to override_os_environ.
214
for var, value in test._original_os_environ.iteritems():
215
# Restore the original value (or delete it if the value has been set to
216
# None in override_os_environ).
217
osutils.set_or_unset_env(var, value)
220
def _clear__type_equality_funcs(test):
221
"""Cleanup bound methods stored on TestCase instances
223
Clear the dict breaking a few (mostly) harmless cycles in the affected
224
unittests released with Python 2.6 and initial Python 2.7 versions.
226
For a few revisions between Python 2.7.1 and Python 2.7.2 that annoyingly
227
shipped in Oneiric, an object with no clear method was used, hence the
228
extra complications, see bug 809048 for details.
230
type_equality_funcs = getattr(test, "_type_equality_funcs", None)
231
if type_equality_funcs is not None:
232
tef_clear = getattr(type_equality_funcs, "clear", None)
233
if tef_clear is None:
234
tef_instance_dict = getattr(type_equality_funcs, "__dict__", None)
235
if tef_instance_dict is not None:
236
tef_clear = tef_instance_dict.clear
237
if tef_clear is not None:
241
class ExtendedTestResult(testtools.TextTestResult):
145
242
"""Accepts, reports and accumulates the results of running tests.
147
244
Compared to the unittest version this class adds support for
218
320
if failed or errored: self.stream.write(", ")
219
321
self.stream.write("known_failure_count=%d" %
220
322
self.known_failure_count)
221
self.stream.writeln(")")
323
self.stream.write(")\n")
223
325
if self.known_failure_count:
224
self.stream.writeln("OK (known_failures=%d)" %
326
self.stream.write("OK (known_failures=%d)\n" %
225
327
self.known_failure_count)
227
self.stream.writeln("OK")
329
self.stream.write("OK\n")
228
330
if self.skip_count > 0:
229
331
skipped = self.skip_count
230
self.stream.writeln('%d test%s skipped' %
332
self.stream.write('%d test%s skipped\n' %
231
333
(skipped, skipped != 1 and "s" or ""))
232
334
if self.unsupported:
233
335
for feature, count in sorted(self.unsupported.items()):
234
self.stream.writeln("Missing feature '%s' skipped %d tests." %
336
self.stream.write("Missing feature '%s' skipped %d tests.\n" %
235
337
(feature, count))
237
339
ok = self.wasStrictlySuccessful()
239
341
ok = self.wasSuccessful()
240
if TestCase._first_thread_leaker_id:
342
if self._first_thread_leaker_id:
241
343
self.stream.write(
242
344
'%s is leaking threads among %d leaking tests.\n' % (
243
TestCase._first_thread_leaker_id,
244
TestCase._leaking_threads_tests))
345
self._first_thread_leaker_id,
346
self._tests_leaking_threads_count))
245
347
# We don't report the main thread as an active one.
246
348
self.stream.write(
247
349
'%d non-main threads were left active in the end.\n'
248
% (TestCase._active_threads - 1))
350
% (len(self._active_threads) - 1))
250
352
def getDescription(self, test):
276
379
def _shortened_test_description(self, test):
278
what = re.sub(r'^bzrlib\.(tests|benchmarks)\.', '', what)
381
what = re.sub(r'^bzrlib\.tests\.', '', what)
384
# GZ 2010-10-04: Cloned tests may end up harmlessly calling this method
385
# multiple times in a row, because the handler is added for
386
# each test but the container list is shared between cases.
387
# See lp:498869 lp:625574 and lp:637725 for background.
388
def _record_traceback_from_test(self, exc_info):
389
"""Store the traceback from passed exc_info tuple till"""
390
self._traceback_from_test = exc_info[2]
281
392
def startTest(self, test):
282
unittest.TestResult.startTest(self, test)
393
super(ExtendedTestResult, self).startTest(test)
283
394
if self.count == 0:
284
395
self.startTests()
285
397
self.report_test_start(test)
286
398
test.number = self.count
287
399
self._recordTestStartTime()
400
# Make testtools cases give us the real traceback on failure
401
addOnException = getattr(test, "addOnException", None)
402
if addOnException is not None:
403
addOnException(self._record_traceback_from_test)
404
# Only check for thread leaks on bzrlib derived test cases
405
if isinstance(test, TestCase):
406
test.addCleanup(self._check_leaked_threads, test)
408
def stopTest(self, test):
409
super(ExtendedTestResult, self).stopTest(test)
410
# Manually break cycles, means touching various private things but hey
411
getDetails = getattr(test, "getDetails", None)
412
if getDetails is not None:
414
_clear__type_equality_funcs(test)
415
self._traceback_from_test = None
289
417
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')
418
self.report_tests_starting()
419
self._active_threads = threading.enumerate()
421
def _check_leaked_threads(self, test):
422
"""See if any threads have leaked since last call
424
A sample of live threads is stored in the _active_threads attribute,
425
when this method runs it compares the current live threads and any not
426
in the previous sample are treated as having leaked.
428
now_active_threads = set(threading.enumerate())
429
threads_leaked = now_active_threads.difference(self._active_threads)
431
self._report_thread_leak(test, threads_leaked, now_active_threads)
432
self._tests_leaking_threads_count += 1
433
if self._first_thread_leaker_id is None:
434
self._first_thread_leaker_id = test.id()
435
self._active_threads = now_active_threads
308
437
def _recordTestStartTime(self):
309
438
"""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:
439
self._start_datetime = self._now()
319
441
def addError(self, test, err):
320
442
"""Tell result that test finished with an error.
356
476
self._formatTime(benchmark_time),
358
478
self.report_success(test)
359
self._cleanupLogFile(test)
360
unittest.TestResult.addSuccess(self, test)
479
super(ExtendedTestResult, self).addSuccess(test)
361
480
test._log_contents = ''
363
482
def addExpectedFailure(self, test, err):
364
483
self.known_failure_count += 1
365
484
self.report_known_failure(test, err)
486
def addUnexpectedSuccess(self, test, details=None):
487
"""Tell result the test unexpectedly passed, counting as a failure
489
When the minimum version of testtools required becomes 0.9.8 this
490
can be updated to use the new handling there.
492
super(ExtendedTestResult, self).addFailure(test, details=details)
493
self.failure_count += 1
494
self.report_unexpected_success(test,
495
"".join(details["reason"].iter_text()))
367
499
def addNotSupported(self, test, feature):
368
500
"""The test will not be run because of a missing feature.
401
538
raise errors.BzrError("Unknown whence %r" % whence)
403
def report_cleaning_up(self):
540
def report_tests_starting(self):
541
"""Display information before the test run begins"""
542
if getattr(sys, 'frozen', None) is None:
543
bzr_path = osutils.realpath(sys.argv[0])
545
bzr_path = sys.executable
547
'bzr selftest: %s\n' % (bzr_path,))
550
bzrlib.__path__[0],))
552
' bzr-%s python-%s %s\n' % (
553
bzrlib.version_string,
554
bzrlib._format_version_tuple(sys.version_info),
555
platform.platform(aliased=1),
557
self.stream.write('\n')
559
def report_test_start(self, test):
560
"""Display information on the test just about to be run"""
562
def _report_thread_leak(self, test, leaked_threads, active_threads):
563
"""Display information on a test that leaked one or more threads"""
564
# GZ 2010-09-09: A leak summary reported separately from the general
565
# thread debugging would be nice. Tests under subunit
566
# need something not using stream, perhaps adding a
567
# testtools details object would be fitting.
568
if 'threads' in selftest_debug_flags:
569
self.stream.write('%s is leaking, active is now %d\n' %
570
(test.id(), len(active_threads)))
406
572
def startTestRun(self):
407
573
self.startTime = time.time()
551
714
return '%s%s' % (indent, err[1])
553
716
def report_error(self, test, err):
554
self.stream.writeln('ERROR %s\n%s'
717
self.stream.write('ERROR %s\n%s\n'
555
718
% (self._testTimeString(test),
556
719
self._error_summary(err)))
558
721
def report_failure(self, test, err):
559
self.stream.writeln(' FAIL %s\n%s'
722
self.stream.write(' FAIL %s\n%s\n'
560
723
% (self._testTimeString(test),
561
724
self._error_summary(err)))
563
726
def report_known_failure(self, test, err):
564
self.stream.writeln('XFAIL %s\n%s'
727
self.stream.write('XFAIL %s\n%s\n'
565
728
% (self._testTimeString(test),
566
729
self._error_summary(err)))
731
def report_unexpected_success(self, test, reason):
732
self.stream.write(' FAIL %s\n%s: %s\n'
733
% (self._testTimeString(test),
734
"Unexpected success. Should have failed",
568
737
def report_success(self, test):
569
self.stream.writeln(' OK %s' % self._testTimeString(test))
738
self.stream.write(' OK %s\n' % self._testTimeString(test))
570
739
for bench_called, stats in getattr(test, '_benchcalls', []):
571
self.stream.writeln('LSProf output for %s(%s, %s)' % bench_called)
740
self.stream.write('LSProf output for %s(%s, %s)\n' % bench_called)
572
741
stats.pprint(file=self.stream)
573
742
# flush the stream so that we get smooth output. This verbose mode is
574
743
# used to show the output in PQM.
575
744
self.stream.flush()
577
746
def report_skip(self, test, reason):
578
self.stream.writeln(' SKIP %s\n%s'
747
self.stream.write(' SKIP %s\n%s\n'
579
748
% (self._testTimeString(test), reason))
581
750
def report_not_applicable(self, test, reason):
582
self.stream.writeln(' N/A %s\n %s'
751
self.stream.write(' N/A %s\n %s\n'
583
752
% (self._testTimeString(test), reason))
585
754
def report_unsupported(self, test, feature):
586
755
"""test cannot be run because feature is missing."""
587
self.stream.writeln("NODEP %s\n The feature '%s' is not available."
756
self.stream.write("NODEP %s\n The feature '%s' is not available.\n"
588
757
%(self._testTimeString(test), feature))
827
1022
self._track_transports()
828
1023
self._track_locks()
829
1024
self._clear_debug_flags()
830
TestCase._active_threads = threading.activeCount()
831
self.addCleanup(self._check_leaked_threads)
1025
# Isolate global verbosity level, to make sure it's reproducible
1026
# between tests. We should get rid of this altogether: bug 656694. --
1028
self.overrideAttr(bzrlib.trace, '_verbosity_level', 0)
1029
self._log_files = set()
1030
# Each key in the ``_counters`` dict holds a value for a different
1031
# counter. When the test ends, addDetail() should be used to output the
1032
# counter values. This happens in install_counter_hook().
1034
if 'config_stats' in selftest_debug_flags:
1035
self._install_config_stats_hooks()
1036
# Do not use i18n for tests (unless the test reverses this)
833
1039
def debug(self):
834
1040
# 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()
1042
# The sys preserved stdin/stdout should allow blackbox tests debugging
1043
pdb.Pdb(stdin=sys.__stdin__, stdout=sys.__stdout__
1044
).set_trace(sys._getframe().f_back)
1046
def discardDetail(self, name):
1047
"""Extend the addDetail, getDetails api so we can remove a detail.
1049
eg. bzr always adds the 'log' detail at startup, but we don't want to
1050
include it for skipped, xfail, etc tests.
1052
It is safe to call this for a detail that doesn't exist, in case this
1053
gets called multiple times.
1055
# We cheat. details is stored in __details which means we shouldn't
1056
# touch it. but getDetails() returns the dict directly, so we can
1058
details = self.getDetails()
1062
def install_counter_hook(self, hooks, name, counter_name=None):
1063
"""Install a counting hook.
1065
Any hook can be counted as long as it doesn't need to return a value.
1067
:param hooks: Where the hook should be installed.
1069
:param name: The hook name that will be counted.
1071
:param counter_name: The counter identifier in ``_counters``, defaults
1074
_counters = self._counters # Avoid closing over self
1075
if counter_name is None:
1077
if _counters.has_key(counter_name):
1078
raise AssertionError('%s is already used as a counter name'
1080
_counters[counter_name] = 0
1081
self.addDetail(counter_name, content.Content(content.UTF8_TEXT,
1082
lambda: ['%d' % (_counters[counter_name],)]))
1083
def increment_counter(*args, **kwargs):
1084
_counters[counter_name] += 1
1085
label = 'count %s calls' % (counter_name,)
1086
hooks.install_named_hook(name, increment_counter, label)
1087
self.addCleanup(hooks.uninstall_named_hook, name, label)
1089
def _install_config_stats_hooks(self):
1090
"""Install config hooks to count hook calls.
1093
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1094
self.install_counter_hook(config.ConfigHooks, hook_name,
1095
'config.%s' % (hook_name,))
1097
# The OldConfigHooks are private and need special handling to protect
1098
# against recursive tests (tests that run other tests), so we just do
1099
# manually what registering them into _builtin_known_hooks will provide
1101
self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1102
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1103
self.install_counter_hook(config.OldConfigHooks, hook_name,
1104
'old_config.%s' % (hook_name,))
853
1106
def _clear_debug_flags(self):
854
1107
"""Prevent externally set debug flags affecting tests.
1322
1588
self.assertEqual(expected_docstring, obj.__doc__)
1590
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1324
1591
def failUnlessExists(self, path):
1592
return self.assertPathExists(path)
1594
def assertPathExists(self, path):
1325
1595
"""Fail unless path or paths, which may be abs or relative, exist."""
1326
1596
if not isinstance(path, basestring):
1328
self.failUnlessExists(p)
1598
self.assertPathExists(p)
1330
self.failUnless(osutils.lexists(path),path+" does not exist")
1600
self.assertTrue(osutils.lexists(path),
1601
path + " does not exist")
1603
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1332
1604
def failIfExists(self, path):
1605
return self.assertPathDoesNotExist(path)
1607
def assertPathDoesNotExist(self, path):
1333
1608
"""Fail if path or paths, which may be abs or relative, exist."""
1334
1609
if not isinstance(path, basestring):
1336
self.failIfExists(p)
1611
self.assertPathDoesNotExist(p)
1338
self.failIf(osutils.lexists(path),path+" exists")
1613
self.assertFalse(osutils.lexists(path),
1340
1616
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1341
1617
"""A helper for callDeprecated and applyDeprecated.
1453
1730
def _startLogFile(self):
1454
"""Send bzr and test log messages to a temporary file.
1456
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
1731
"""Setup a in-memory target for bzr and testcase log messages"""
1732
pseudo_log_file = StringIO()
1733
def _get_log_contents_for_weird_testtools_api():
1734
return [pseudo_log_file.getvalue().decode(
1735
"utf-8", "replace").encode("utf-8")]
1736
self.addDetail("log", content.Content(content.ContentType("text",
1737
"plain", {"charset": "utf8"}),
1738
_get_log_contents_for_weird_testtools_api))
1739
self._log_file = pseudo_log_file
1740
self._log_memento = trace.push_log_file(self._log_file)
1462
1741
self.addCleanup(self._finishLogFile)
1464
1743
def _finishLogFile(self):
1465
"""Finished with the log file.
1467
Close the file and delete it, unless setKeepLogfile was called.
1469
if bzrlib.trace._trace_file:
1744
"""Flush and dereference the in-memory log for this testcase"""
1745
if trace._trace_file:
1470
1746
# 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)
1747
trace._trace_file.flush()
1748
trace.pop_log_file(self._log_memento)
1749
# The logging module now tracks references for cleanup so discard ours
1750
del self._log_memento
1476
1752
def thisFailsStrictLockCheck(self):
1477
1753
"""It is known that this test would fail with -Dstrict_locks.
1513
1784
setattr(obj, attr_name, new)
1787
def overrideEnv(self, name, new):
1788
"""Set an environment variable, and reset it after the test.
1790
:param name: The environment variable name.
1792
:param new: The value to set the variable to. If None, the
1793
variable is deleted from the environment.
1795
:returns: The actual variable value.
1797
value = osutils.set_or_unset_env(name, new)
1798
self.addCleanup(osutils.set_or_unset_env, name, value)
1801
def recordCalls(self, obj, attr_name):
1802
"""Monkeypatch in a wrapper that will record calls.
1804
The monkeypatch is automatically removed when the test concludes.
1806
:param obj: The namespace holding the reference to be replaced;
1807
typically a module, class, or object.
1808
:param attr_name: A string for the name of the attribute to
1810
:returns: A list that will be extended with one item every time the
1811
function is called, with a tuple of (args, kwargs).
1815
def decorator(*args, **kwargs):
1816
calls.append((args, kwargs))
1817
return orig(*args, **kwargs)
1818
orig = self.overrideAttr(obj, attr_name, decorator)
1516
1821
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)
1822
for name, value in isolated_environ.iteritems():
1823
self.overrideEnv(name, value)
1578
1825
def _restoreHooks(self):
1579
1826
for klass, (name, hooks) in self._preserved_hooks.items():
1580
1827
setattr(klass, name, hooks)
1828
self._preserved_hooks.clear()
1829
bzrlib.hooks._lazy_hooks = self._preserved_lazy_hooks
1830
self._preserved_lazy_hooks.clear()
1582
1832
def knownFailure(self, reason):
1583
"""This test has failed for some known reason."""
1584
raise KnownFailure(reason)
1833
"""Declare that this test fails for a known reason
1835
Tests that are known to fail should generally be using expectedFailure
1836
with an appropriate reverse assertion if a change could cause the test
1837
to start passing. Conversely if the test has no immediate prospect of
1838
succeeding then using skip is more suitable.
1840
When this method is called while an exception is being handled, that
1841
traceback will be used, otherwise a new exception will be thrown to
1842
provide one but won't be reported.
1844
self._add_reason(reason)
1846
exc_info = sys.exc_info()
1847
if exc_info != (None, None, None):
1848
self._report_traceback(exc_info)
1851
raise self.failureException(reason)
1852
except self.failureException:
1853
exc_info = sys.exc_info()
1854
# GZ 02-08-2011: Maybe cleanup this err.exc_info attribute too?
1855
raise testtools.testcase._ExpectedFailure(exc_info)
1859
def _suppress_log(self):
1860
"""Remove the log info from details."""
1861
self.discardDetail('log')
1586
1863
def _do_skip(self, result, reason):
1864
self._suppress_log()
1587
1865
addSkip = getattr(result, 'addSkip', None)
1588
1866
if not callable(addSkip):
1589
1867
result.addSuccess(result)
1645
1946
self._benchtime += time.time() - start
1647
1948
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
1951
def get_log(self):
1730
1952
"""Get a unicode string containing the log from bzrlib.trace.
1945
2168
variables. A value of None will unset the env variable.
1946
2169
The values must be strings. The change will only occur in the
1947
2170
child, so you don't need to fix the environment after running.
1948
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
2171
:param skip_if_plan_to_signal: raise TestSkipped when true and system
2172
doesn't support signalling subprocesses.
1950
2173
:param allow_plugins: If False (default) pass --no-plugins to bzr.
2174
:param stderr: file to use for the subprocess's stderr. Valid values
2175
are those valid for the stderr argument of `subprocess.Popen`.
2176
Default value is ``subprocess.PIPE``.
1952
2178
:returns: Popen object for the started process.
1954
2180
if skip_if_plan_to_signal:
1955
if not getattr(os, 'kill', None):
1956
raise TestSkipped("os.kill not available.")
2181
if os.name != "posix":
2182
raise TestSkipped("Sending signals not supported")
1958
2184
if env_changes is None:
1959
2185
env_changes = {}
2186
# Because $HOME is set to a tempdir for the context of a test, modules
2187
# installed in the user dir will not be found unless $PYTHONUSERBASE
2188
# gets set to the computed directory of this parent process.
2189
if site.USER_BASE is not None:
2190
env_changes["PYTHONUSERBASE"] = site.USER_BASE
1962
2193
def cleanup_environment():
2233
def _add_subprocess_log(self, log_file_path):
2234
if len(self._log_files) == 0:
2235
# Register an addCleanup func. We do this on the first call to
2236
# _add_subprocess_log rather than in TestCase.setUp so that this
2237
# addCleanup is registered after any cleanups for tempdirs that
2238
# subclasses might create, which will probably remove the log file
2240
self.addCleanup(self._subprocess_log_cleanup)
2241
# self._log_files is a set, so if a log file is reused we won't grab it
2243
self._log_files.add(log_file_path)
2245
def _subprocess_log_cleanup(self):
2246
for count, log_file_path in enumerate(self._log_files):
2247
# We use buffer_now=True to avoid holding the file open beyond
2248
# the life of this function, which might interfere with e.g.
2249
# cleaning tempdirs on Windows.
2250
# XXX: Testtools 0.9.5 doesn't have the content_from_file helper
2251
#detail_content = content.content_from_file(
2252
# log_file_path, buffer_now=True)
2253
with open(log_file_path, 'rb') as log_file:
2254
log_file_bytes = log_file.read()
2255
detail_content = content.Content(content.ContentType("text",
2256
"plain", {"charset": "utf8"}), lambda: [log_file_bytes])
2257
self.addDetail("start_bzr_subprocess-log-%d" % (count,),
1997
2260
def _popen(self, *args, **kwargs):
1998
2261
"""Place a call to Popen.
2000
2263
Allows tests to override this method to intercept the calls made to
2001
2264
Popen for introspection.
2003
return Popen(*args, **kwargs)
2266
return subprocess.Popen(*args, **kwargs)
2005
2268
def get_source_path(self):
2006
2269
"""Return the path of the directory containing bzrlib."""
2036
2299
if retcode is not None and retcode != process.returncode:
2037
2300
if process_args is None:
2038
2301
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)
2302
trace.mutter('Output of bzr %s:\n%s', process_args, out)
2303
trace.mutter('Error for bzr %s:\n%s', process_args, err)
2041
2304
self.fail('Command bzr %s failed with retcode %s != %s'
2042
2305
% (process_args, retcode, process.returncode))
2043
2306
return [out, err]
2045
def check_inventory_shape(self, inv, shape):
2046
"""Compare an inventory to a list of expected names.
2308
def check_tree_shape(self, tree, shape):
2309
"""Compare a tree to a list of expected names.
2048
2311
Fail if they are not precisely equal.
2051
2314
shape = list(shape) # copy
2052
for path, ie in inv.entries():
2315
for path, ie in tree.iter_entries_by_dir():
2053
2316
name = path.replace('\\', '/')
2054
2317
if ie.kind == 'directory':
2055
2318
name = name + '/'
2320
pass # ignore root entry
2057
2322
shape.remove(name)
2059
2324
extras.append(name)
2149
2416
class TestCaseWithMemoryTransport(TestCase):
2150
2417
"""Common test class for tests that do not need disk resources.
2152
Tests that need disk resources should derive from TestCaseWithTransport.
2419
Tests that need disk resources should derive from TestCaseInTempDir
2420
orTestCaseWithTransport.
2154
2422
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
2156
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
2424
For TestCaseWithMemoryTransport the ``test_home_dir`` is set to the name of
2157
2425
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
2426
is preserved. ``test_dir`` is set to the TEST_ROOT, as is cwd, because they
2427
must exist. However, TestCaseWithMemoryTransport does not offer local file
2428
defaults for the transport in tests, nor does it obey the command line
2161
2429
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.
2432
:cvar TEST_ROOT: Directory containing all temporary directories, plus a
2433
``.bzr`` directory that stops us ascending higher into the filesystem.
2168
2436
TEST_ROOT = None
2375
2654
self.test_home_dir = self.test_dir + "/MemoryTransportMissingHomeDir"
2376
2655
self.permit_dir(self.test_dir)
2378
def make_branch(self, relpath, format=None):
2657
def make_branch(self, relpath, format=None, name=None):
2379
2658
"""Create a branch on the transport at relpath."""
2380
2659
repo = self.make_repository(relpath, format=format)
2381
return repo.bzrdir.create_branch()
2660
return repo.bzrdir.create_branch(append_revisions_only=False,
2663
def get_default_format(self):
2666
def resolve_format(self, format):
2667
"""Resolve an object to a ControlDir format object.
2669
The initial format object can either already be
2670
a ControlDirFormat, None (for the default format),
2671
or a string with the name of the control dir format.
2673
:param format: Object to resolve
2674
:return A ControlDirFormat instance
2677
format = self.get_default_format()
2678
if isinstance(format, basestring):
2679
format = controldir.format_registry.make_bzrdir(format)
2383
2682
def make_bzrdir(self, relpath, format=None):
2385
2684
# might be a relative or absolute path
2386
2685
maybe_a_url = self.get_url(relpath)
2387
2686
segments = maybe_a_url.rsplit('/', 1)
2388
t = get_transport(maybe_a_url)
2687
t = _mod_transport.get_transport(maybe_a_url)
2389
2688
if len(segments) > 1 and segments[-1] not in ('', '.'):
2390
2689
t.ensure_base()
2393
if isinstance(format, basestring):
2394
format = bzrdir.format_registry.make_bzrdir(format)
2690
format = self.resolve_format(format)
2395
2691
return format.initialize_on_transport(t)
2396
2692
except errors.UninitializableFormat:
2397
2693
raise TestSkipped("Format %s is not initializable." % format)
2399
def make_repository(self, relpath, shared=False, format=None):
2695
def make_repository(self, relpath, shared=None, format=None):
2400
2696
"""Create a repository on our default transport at relpath.
2402
2698
Note that relpath must be a relative path, not a full url.
3102
3424
"""A decorator which excludes test matching an exclude pattern."""
3104
3426
def __init__(self, suite, exclude_pattern):
3105
TestDecorator.__init__(self, suite)
3106
self.exclude_pattern = exclude_pattern
3107
self.excluded = False
3111
return iter(self._tests)
3112
self.excluded = True
3113
suite = exclude_tests_by_re(self, self.exclude_pattern)
3115
self.addTests(suite)
3116
return iter(self._tests)
3427
super(ExcludeDecorator, self).__init__(
3428
exclude_tests_by_re(suite, exclude_pattern))
3119
3431
class FilterTestsDecorator(TestDecorator):
3120
3432
"""A decorator which filters tests to those matching a pattern."""
3122
3434
def __init__(self, suite, pattern):
3123
TestDecorator.__init__(self, suite)
3124
self.pattern = pattern
3125
self.filtered = False
3129
return iter(self._tests)
3130
self.filtered = True
3131
suite = filter_suite_by_re(self, self.pattern)
3133
self.addTests(suite)
3134
return iter(self._tests)
3435
super(FilterTestsDecorator, self).__init__(
3436
filter_suite_by_re(suite, pattern))
3137
3439
class RandomDecorator(TestDecorator):
3138
3440
"""A decorator which randomises the order of its tests."""
3140
3442
def __init__(self, suite, random_seed, stream):
3141
TestDecorator.__init__(self, suite)
3142
self.random_seed = random_seed
3143
self.randomised = False
3144
self.stream = stream
3148
return iter(self._tests)
3149
self.randomised = True
3150
self.stream.write("Randomizing test order using seed %s\n\n" %
3151
(self.actual_seed()))
3443
random_seed = self.actual_seed(random_seed)
3444
stream.write("Randomizing test order using seed %s\n\n" %
3152
3446
# Initialise the random number generator.
3153
random.seed(self.actual_seed())
3154
suite = randomize_suite(self)
3156
self.addTests(suite)
3157
return iter(self._tests)
3447
random.seed(random_seed)
3448
super(RandomDecorator, self).__init__(randomize_suite(suite))
3159
def actual_seed(self):
3160
if self.random_seed == "now":
3451
def actual_seed(seed):
3161
3453
# We convert the seed to a long to make it reuseable across
3162
3454
# invocations (because the user can reenter it).
3163
self.random_seed = long(time.time())
3455
return long(time.time())
3165
3457
# Convert the seed to a long if we can
3167
self.random_seed = long(self.random_seed)
3460
except (TypeError, ValueError):
3170
return self.random_seed
3173
3465
class TestFirstDecorator(TestDecorator):
3174
3466
"""A decorator which moves named tests to the front."""
3176
3468
def __init__(self, suite, pattern):
3177
TestDecorator.__init__(self, suite)
3178
self.pattern = pattern
3179
self.filtered = False
3183
return iter(self._tests)
3184
self.filtered = True
3185
suites = split_suite_by_re(self, self.pattern)
3187
self.addTests(suites)
3188
return iter(self._tests)
3469
super(TestFirstDecorator, self).__init__()
3470
self.addTests(split_suite_by_re(suite, pattern))
3191
3473
def partition_tests(suite, count):
3192
3474
"""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)
3475
# This just assigns tests in a round-robin fashion. On one hand this
3476
# splits up blocks of related tests that might run faster if they shared
3477
# resources, but on the other it avoids assigning blocks of slow tests to
3478
# just one partition. So the slowest partition shouldn't be much slower
3480
partitions = [list() for i in range(count)]
3481
tests = iter_suite_tests(suite)
3482
for partition, test in itertools.izip(itertools.cycle(partitions), tests):
3483
partition.append(test)
3204
3487
def workaround_zealous_crypto_random():
3235
3518
ProtocolTestCase.run(self, result)
3237
os.waitpid(self.pid, 0)
3520
pid, status = os.waitpid(self.pid, 0)
3521
# GZ 2011-10-18: If status is nonzero, should report to the result
3522
# that something went wrong.
3239
3524
test_blocks = partition_tests(suite, concurrency)
3525
# Clear the tests from the original suite so it doesn't keep them alive
3526
suite._tests[:] = []
3240
3527
for process_tests in test_blocks:
3241
process_suite = TestSuite()
3242
process_suite.addTests(process_tests)
3528
process_suite = TestUtil.TestSuite(process_tests)
3529
# Also clear each split list so new suite has only reference
3530
process_tests[:] = []
3243
3531
c2pread, c2pwrite = os.pipe()
3244
3532
pid = os.fork()
3246
workaround_zealous_crypto_random()
3535
stream = os.fdopen(c2pwrite, 'wb', 1)
3536
workaround_zealous_crypto_random()
3248
3537
os.close(c2pread)
3249
3538
# Leave stderr and stdout open so we can see test noise
3250
3539
# Close stdin so that the child goes away if it decides to
3251
3540
# read from stdin (otherwise its a roulette to see what
3252
3541
# child actually gets keystrokes for pdb etc).
3253
3542
sys.stdin.close()
3255
stream = os.fdopen(c2pwrite, 'wb', 1)
3256
3543
subunit_result = AutoTimingTestResultDecorator(
3257
TestProtocolClient(stream))
3544
SubUnitBzrProtocolClient(stream))
3258
3545
process_suite.run(subunit_result)
3547
# Try and report traceback on stream, but exit with error even
3548
# if stream couldn't be created or something else goes wrong.
3549
# The traceback is formatted to a string and written in one go
3550
# to avoid interleaving lines from multiple failing children.
3552
stream.write(traceback.format_exc())
3262
3557
os.close(c2pwrite)
3263
3558
stream = os.fdopen(c2pread, 'rb', 1)
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):
3622
class ProfileResult(testtools.ExtendedToOriginalDecorator):
3358
3623
"""Generate profiling data for all activity between start and success.
3360
3625
The profile data is appended to the test's _benchcalls attribute and can
3686
3964
'bzrlib.tests.test_commit_merge',
3687
3965
'bzrlib.tests.test_config',
3688
3966
'bzrlib.tests.test_conflicts',
3967
'bzrlib.tests.test_controldir',
3689
3968
'bzrlib.tests.test_counted_lock',
3690
3969
'bzrlib.tests.test_crash',
3691
3970
'bzrlib.tests.test_decorators',
3692
3971
'bzrlib.tests.test_delta',
3693
3972
'bzrlib.tests.test_debug',
3694
'bzrlib.tests.test_deprecated_graph',
3695
3973
'bzrlib.tests.test_diff',
3696
3974
'bzrlib.tests.test_directory_service',
3697
3975
'bzrlib.tests.test_dirstate',
3698
3976
'bzrlib.tests.test_email_message',
3699
3977
'bzrlib.tests.test_eol_filters',
3700
3978
'bzrlib.tests.test_errors',
3979
'bzrlib.tests.test_estimate_compressed_size',
3701
3980
'bzrlib.tests.test_export',
3981
'bzrlib.tests.test_export_pot',
3702
3982
'bzrlib.tests.test_extract',
3983
'bzrlib.tests.test_features',
3703
3984
'bzrlib.tests.test_fetch',
3985
'bzrlib.tests.test_fixtures',
3704
3986
'bzrlib.tests.test_fifo_cache',
3705
3987
'bzrlib.tests.test_filters',
3988
'bzrlib.tests.test_filter_tree',
3706
3989
'bzrlib.tests.test_ftp_transport',
3707
3990
'bzrlib.tests.test_foreign',
3708
3991
'bzrlib.tests.test_generate_docs',
4102
4426
if test_id != None:
4103
4427
ui.ui_factory.clear_term()
4104
4428
sys.stderr.write('\nWhile running: %s\n' % (test_id,))
4429
# Ugly, but the last thing we want here is fail, so bear with it.
4430
printable_e = str(e).decode(osutils.get_user_encoding(), 'replace'
4431
).encode('ascii', 'replace')
4105
4432
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')
4433
% (os.path.basename(dirname), printable_e))
4268
4436
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
4469
# Only define SubUnitBzrRunner if subunit is available.
4455
4471
from subunit import TestProtocolClient
4456
4472
from subunit.test_results import AutoTimingTestResultDecorator
4473
class SubUnitBzrProtocolClient(TestProtocolClient):
4475
def stopTest(self, test):
4476
super(SubUnitBzrProtocolClient, self).stopTest(test)
4477
_clear__type_equality_funcs(test)
4479
def addSuccess(self, test, details=None):
4480
# The subunit client always includes the details in the subunit
4481
# stream, but we don't want to include it in ours.
4482
if details is not None and 'log' in details:
4484
return super(SubUnitBzrProtocolClient, self).addSuccess(
4457
4487
class SubUnitBzrRunner(TextTestRunner):
4458
4488
def run(self, test):
4459
4489
result = AutoTimingTestResultDecorator(
4460
TestProtocolClient(self.stream))
4490
SubUnitBzrProtocolClient(self.stream))
4461
4491
test.run(result)
4463
4493
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()
4497
# API compatibility for old plugins; see bug 892622.
4500
'HTTPServerFeature',
4501
'ModuleAvailableFeature',
4502
'HTTPSServerFeature', 'SymlinkFeature', 'HardlinkFeature',
4503
'OsFifoFeature', 'UnicodeFilenameFeature',
4504
'ByteStringNamedFilesystem', 'UTF8Filesystem',
4505
'BreakinFeature', 'CaseInsCasePresFilenameFeature',
4506
'CaseInsensitiveFilesystemFeature', 'case_sensitive_filesystem_feature',
4507
'posix_permissions_feature',
4509
globals()[name] = _CompatabilityThunkFeature(
4510
symbol_versioning.deprecated_in((2, 5, 0)),
4511
'bzrlib.tests', name,
4512
name, 'bzrlib.tests.features')
4515
for (old_name, new_name) in [
4516
('UnicodeFilename', 'UnicodeFilenameFeature'),
4518
globals()[name] = _CompatabilityThunkFeature(
4519
symbol_versioning.deprecated_in((2, 5, 0)),
4520
'bzrlib.tests', old_name,
4521
new_name, 'bzrlib.tests.features')
added
removed
bzrlib/tests/__init__.py
