210
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:
213
241
class ExtendedTestResult(testtools.TextTestResult):
214
242
"""Accepts, reports and accumulates the results of running tests.
377
405
if isinstance(test, TestCase):
378
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
380
417
def startTests(self):
381
418
self.report_tests_starting()
382
419
self._active_threads = threading.enumerate()
384
def stopTest(self, test):
385
self._traceback_from_test = None
387
421
def _check_leaked_threads(self, test):
388
422
"""See if any threads have leaked since last call
449
483
self.known_failure_count += 1
450
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()))
452
499
def addNotSupported(self, test, feature):
453
500
"""The test will not be run because of a missing feature.
936
1000
def setUp(self):
937
1001
super(TestCase, self).setUp()
1003
timeout = config.GlobalStack().get('selftest.timeout')
1005
timeout_fixture = fixtures.TimeoutFixture(timeout)
1006
timeout_fixture.setUp()
1007
self.addCleanup(timeout_fixture.cleanUp)
938
1009
for feature in getattr(self, '_test_needs_features', []):
939
1010
self.requireFeature(feature)
940
self._log_contents = None
941
self.addDetail("log", content.Content(content.ContentType("text",
942
"plain", {"charset": "utf8"}),
943
lambda:[self._get_log(keep_log_file=True)]))
944
1011
self._cleanEnvironment()
1013
if bzrlib.global_state is not None:
1014
self.overrideAttr(bzrlib.global_state, 'cmdline_overrides',
1015
config.CommandLineStore())
945
1017
self._silenceUI()
946
1018
self._startLogFile()
947
1019
self._benchcalls = []
954
1026
# between tests. We should get rid of this altogether: bug 656694. --
956
1028
self.overrideAttr(bzrlib.trace, '_verbosity_level', 0)
1029
# Isolate config option expansion until its default value for bzrlib is
1030
# settled on or a the FIXME associated with _get_expand_default_value
1031
# is addressed -- vila 20110219
1032
self.overrideAttr(config, '_expand_default_value', None)
1033
self._log_files = set()
1034
# Each key in the ``_counters`` dict holds a value for a different
1035
# counter. When the test ends, addDetail() should be used to output the
1036
# counter values. This happens in install_counter_hook().
1038
if 'config_stats' in selftest_debug_flags:
1039
self._install_config_stats_hooks()
1040
# Do not use i18n for tests (unless the test reverses this)
958
1043
def debug(self):
959
1044
# debug a frame up.
961
pdb.Pdb().set_trace(sys._getframe().f_back)
1046
# The sys preserved stdin/stdout should allow blackbox tests debugging
1047
pdb.Pdb(stdin=sys.__stdin__, stdout=sys.__stdout__
1048
).set_trace(sys._getframe().f_back)
963
1050
def discardDetail(self, name):
964
1051
"""Extend the addDetail, getDetails api so we can remove a detail.
976
1063
if name in details:
977
1064
del details[name]
1066
def install_counter_hook(self, hooks, name, counter_name=None):
1067
"""Install a counting hook.
1069
Any hook can be counted as long as it doesn't need to return a value.
1071
:param hooks: Where the hook should be installed.
1073
:param name: The hook name that will be counted.
1075
:param counter_name: The counter identifier in ``_counters``, defaults
1078
_counters = self._counters # Avoid closing over self
1079
if counter_name is None:
1081
if _counters.has_key(counter_name):
1082
raise AssertionError('%s is already used as a counter name'
1084
_counters[counter_name] = 0
1085
self.addDetail(counter_name, content.Content(content.UTF8_TEXT,
1086
lambda: ['%d' % (_counters[counter_name],)]))
1087
def increment_counter(*args, **kwargs):
1088
_counters[counter_name] += 1
1089
label = 'count %s calls' % (counter_name,)
1090
hooks.install_named_hook(name, increment_counter, label)
1091
self.addCleanup(hooks.uninstall_named_hook, name, label)
1093
def _install_config_stats_hooks(self):
1094
"""Install config hooks to count hook calls.
1097
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1098
self.install_counter_hook(config.ConfigHooks, hook_name,
1099
'config.%s' % (hook_name,))
1101
# The OldConfigHooks are private and need special handling to protect
1102
# against recursive tests (tests that run other tests), so we just do
1103
# manually what registering them into _builtin_known_hooks will provide
1105
self.overrideAttr(config, 'OldConfigHooks', config._OldConfigHooks())
1106
for hook_name in ('get', 'set', 'remove', 'load', 'save'):
1107
self.install_counter_hook(config.OldConfigHooks, hook_name,
1108
'old_config.%s' % (hook_name,))
979
1110
def _clear_debug_flags(self):
980
1111
"""Prevent externally set debug flags affecting tests.
1033
1166
# break some locks on purpose and should be taken into account by
1034
1167
# considering that breaking a lock is just a dirty way of releasing it.
1035
1168
if len(acquired_locks) != (len(released_locks) + len(broken_locks)):
1036
message = ('Different number of acquired and '
1037
'released or broken locks. (%s, %s + %s)' %
1038
(acquired_locks, released_locks, broken_locks))
1170
'Different number of acquired and '
1171
'released or broken locks.\n'
1175
(acquired_locks, released_locks, broken_locks))
1039
1176
if not self._lock_check_thorough:
1040
1177
# Rather than fail, just warn
1041
1178
print "Broken test %s: %s" % (self, message)
1261
1398
'st_mtime did not match')
1262
1399
self.assertEqual(expected.st_ctime, actual.st_ctime,
1263
1400
'st_ctime did not match')
1264
if sys.platform != 'win32':
1401
if sys.platform == 'win32':
1265
1402
# On Win32 both 'dev' and 'ino' cannot be trusted. In python2.4 it
1266
1403
# is 'dev' that varies, in python 2.5 (6?) it is st_ino that is
1267
# odd. Regardless we shouldn't actually try to assert anything
1268
# about their values
1404
# odd. We just force it to always be 0 to avoid any problems.
1405
self.assertEqual(0, expected.st_dev)
1406
self.assertEqual(0, actual.st_dev)
1407
self.assertEqual(0, expected.st_ino)
1408
self.assertEqual(0, actual.st_ino)
1269
1410
self.assertEqual(expected.st_dev, actual.st_dev,
1270
1411
'st_dev did not match')
1271
1412
self.assertEqual(expected.st_ino, actual.st_ino,
1280
1421
length, len(obj_with_len), obj_with_len))
1282
1423
def assertLogsError(self, exception_class, func, *args, **kwargs):
1283
"""Assert that func(*args, **kwargs) quietly logs a specific exception.
1424
"""Assert that `func(*args, **kwargs)` quietly logs a specific error.
1286
1427
orig_log_exception_quietly = trace.log_exception_quietly
1289
1430
orig_log_exception_quietly()
1290
captured.append(sys.exc_info())
1431
captured.append(sys.exc_info()[1])
1291
1432
trace.log_exception_quietly = capture
1292
1433
func(*args, **kwargs)
1294
1435
trace.log_exception_quietly = orig_log_exception_quietly
1295
1436
self.assertLength(1, captured)
1296
err = captured[0][1]
1297
1438
self.assertIsInstance(err, exception_class)
1451
1592
self.assertEqual(expected_docstring, obj.__doc__)
1594
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1453
1595
def failUnlessExists(self, path):
1596
return self.assertPathExists(path)
1598
def assertPathExists(self, path):
1454
1599
"""Fail unless path or paths, which may be abs or relative, exist."""
1455
1600
if not isinstance(path, basestring):
1457
self.failUnlessExists(p)
1602
self.assertPathExists(p)
1459
self.failUnless(osutils.lexists(path),path+" does not exist")
1604
self.assertTrue(osutils.lexists(path),
1605
path + " does not exist")
1607
@symbol_versioning.deprecated_method(symbol_versioning.deprecated_in((2, 4)))
1461
1608
def failIfExists(self, path):
1609
return self.assertPathDoesNotExist(path)
1611
def assertPathDoesNotExist(self, path):
1462
1612
"""Fail if path or paths, which may be abs or relative, exist."""
1463
1613
if not isinstance(path, basestring):
1465
self.failIfExists(p)
1615
self.assertPathDoesNotExist(p)
1467
self.failIf(osutils.lexists(path),path+" exists")
1617
self.assertFalse(osutils.lexists(path),
1469
1620
def _capture_deprecation_warnings(self, a_callable, *args, **kwargs):
1470
1621
"""A helper for callDeprecated and applyDeprecated.
1582
1734
def _startLogFile(self):
1583
"""Send bzr and test log messages to a temporary file.
1585
The file is removed as the test is torn down.
1587
self._log_file = StringIO()
1735
"""Setup a in-memory target for bzr and testcase log messages"""
1736
pseudo_log_file = StringIO()
1737
def _get_log_contents_for_weird_testtools_api():
1738
return [pseudo_log_file.getvalue().decode(
1739
"utf-8", "replace").encode("utf-8")]
1740
self.addDetail("log", content.Content(content.ContentType("text",
1741
"plain", {"charset": "utf8"}),
1742
_get_log_contents_for_weird_testtools_api))
1743
self._log_file = pseudo_log_file
1588
1744
self._log_memento = trace.push_log_file(self._log_file)
1589
1745
self.addCleanup(self._finishLogFile)
1591
1747
def _finishLogFile(self):
1592
"""Finished with the log file.
1594
Close the file and delete it, unless setKeepLogfile was called.
1748
"""Flush and dereference the in-memory log for this testcase"""
1596
1749
if trace._trace_file:
1597
1750
# flush the log file, to get all content
1598
1751
trace._trace_file.flush()
1599
1752
trace.pop_log_file(self._log_memento)
1600
# Cache the log result and delete the file on disk
1601
self._get_log(False)
1753
# The logging module now tracks references for cleanup so discard ours
1754
del self._log_memento
1603
1756
def thisFailsStrictLockCheck(self):
1604
1757
"""It is known that this test would fail with -Dstrict_locks.
1646
1802
self.addCleanup(osutils.set_or_unset_env, name, value)
1805
def recordCalls(self, obj, attr_name):
1806
"""Monkeypatch in a wrapper that will record calls.
1808
The monkeypatch is automatically removed when the test concludes.
1810
:param obj: The namespace holding the reference to be replaced;
1811
typically a module, class, or object.
1812
:param attr_name: A string for the name of the attribute to
1814
:returns: A list that will be extended with one item every time the
1815
function is called, with a tuple of (args, kwargs).
1819
def decorator(*args, **kwargs):
1820
calls.append((args, kwargs))
1821
return orig(*args, **kwargs)
1822
orig = self.overrideAttr(obj, attr_name, decorator)
1649
1825
def _cleanEnvironment(self):
1650
1826
for name, value in isolated_environ.iteritems():
1651
1827
self.overrideEnv(name, value)
1653
1829
def _restoreHooks(self):
1654
1830
for klass, (name, hooks) in self._preserved_hooks.items():
1655
1831
setattr(klass, name, hooks)
1832
self._preserved_hooks.clear()
1833
bzrlib.hooks._lazy_hooks = self._preserved_lazy_hooks
1834
self._preserved_lazy_hooks.clear()
1657
1836
def knownFailure(self, reason):
1658
"""This test has failed for some known reason."""
1659
raise KnownFailure(reason)
1837
"""Declare that this test fails for a known reason
1839
Tests that are known to fail should generally be using expectedFailure
1840
with an appropriate reverse assertion if a change could cause the test
1841
to start passing. Conversely if the test has no immediate prospect of
1842
succeeding then using skip is more suitable.
1844
When this method is called while an exception is being handled, that
1845
traceback will be used, otherwise a new exception will be thrown to
1846
provide one but won't be reported.
1848
self._add_reason(reason)
1850
exc_info = sys.exc_info()
1851
if exc_info != (None, None, None):
1852
self._report_traceback(exc_info)
1855
raise self.failureException(reason)
1856
except self.failureException:
1857
exc_info = sys.exc_info()
1858
# GZ 02-08-2011: Maybe cleanup this err.exc_info attribute too?
1859
raise testtools.testcase._ExpectedFailure(exc_info)
1661
1863
def _suppress_log(self):
1662
1864
"""Remove the log info from details."""
1750
1952
def log(self, *args):
1751
1953
trace.mutter(*args)
1753
def _get_log(self, keep_log_file=False):
1754
"""Internal helper to get the log from bzrlib.trace for this test.
1756
Please use self.getDetails, or self.get_log to access this in test case
1759
:param keep_log_file: When True, if the log is still a file on disk
1760
leave it as a file on disk. When False, if the log is still a file
1761
on disk, the log file is deleted and the log preserved as
1763
:return: A string containing the log.
1765
if self._log_contents is not None:
1767
self._log_contents.decode('utf8')
1768
except UnicodeDecodeError:
1769
unicodestr = self._log_contents.decode('utf8', 'replace')
1770
self._log_contents = unicodestr.encode('utf8')
1771
return self._log_contents
1772
if self._log_file is not None:
1773
log_contents = self._log_file.getvalue()
1775
log_contents.decode('utf8')
1776
except UnicodeDecodeError:
1777
unicodestr = log_contents.decode('utf8', 'replace')
1778
log_contents = unicodestr.encode('utf8')
1779
if not keep_log_file:
1780
self._log_file = None
1781
# Permit multiple calls to get_log until we clean it up in
1783
self._log_contents = log_contents
1786
return "No log file content."
1788
1955
def get_log(self):
1789
1956
"""Get a unicode string containing the log from bzrlib.trace.
2237
def _add_subprocess_log(self, log_file_path):
2238
if len(self._log_files) == 0:
2239
# Register an addCleanup func. We do this on the first call to
2240
# _add_subprocess_log rather than in TestCase.setUp so that this
2241
# addCleanup is registered after any cleanups for tempdirs that
2242
# subclasses might create, which will probably remove the log file
2244
self.addCleanup(self._subprocess_log_cleanup)
2245
# self._log_files is a set, so if a log file is reused we won't grab it
2247
self._log_files.add(log_file_path)
2249
def _subprocess_log_cleanup(self):
2250
for count, log_file_path in enumerate(self._log_files):
2251
# We use buffer_now=True to avoid holding the file open beyond
2252
# the life of this function, which might interfere with e.g.
2253
# cleaning tempdirs on Windows.
2254
# XXX: Testtools 0.9.5 doesn't have the content_from_file helper
2255
#detail_content = content.content_from_file(
2256
# log_file_path, buffer_now=True)
2257
with open(log_file_path, 'rb') as log_file:
2258
log_file_bytes = log_file.read()
2259
detail_content = content.Content(content.ContentType("text",
2260
"plain", {"charset": "utf8"}), lambda: [log_file_bytes])
2261
self.addDetail("start_bzr_subprocess-log-%d" % (count,),
2059
2264
def _popen(self, *args, **kwargs):
2060
2265
"""Place a call to Popen.
2104
2309
% (process_args, retcode, process.returncode))
2105
2310
return [out, err]
2107
def check_inventory_shape(self, inv, shape):
2108
"""Compare an inventory to a list of expected names.
2312
def check_tree_shape(self, tree, shape):
2313
"""Compare a tree to a list of expected names.
2110
2315
Fail if they are not precisely equal.
2113
2318
shape = list(shape) # copy
2114
for path, ie in inv.entries():
2319
for path, ie in tree.iter_entries_by_dir():
2115
2320
name = path.replace('\\', '/')
2116
2321
if ie.kind == 'directory':
2117
2322
name = name + '/'
2324
pass # ignore root entry
2119
2326
shape.remove(name)
2121
2328
extras.append(name)
2211
2420
class TestCaseWithMemoryTransport(TestCase):
2212
2421
"""Common test class for tests that do not need disk resources.
2214
Tests that need disk resources should derive from TestCaseWithTransport.
2423
Tests that need disk resources should derive from TestCaseInTempDir
2424
orTestCaseWithTransport.
2216
2426
TestCaseWithMemoryTransport sets the TEST_ROOT variable for all bzr tests.
2218
For TestCaseWithMemoryTransport the test_home_dir is set to the name of
2428
For TestCaseWithMemoryTransport the ``test_home_dir`` is set to the name of
2219
2429
a directory which does not exist. This serves to help ensure test isolation
2220
is preserved. test_dir is set to the TEST_ROOT, as is cwd, because they
2221
must exist. However, TestCaseWithMemoryTransport does not offer local
2222
file defaults for the transport in tests, nor does it obey the command line
2430
is preserved. ``test_dir`` is set to the TEST_ROOT, as is cwd, because they
2431
must exist. However, TestCaseWithMemoryTransport does not offer local file
2432
defaults for the transport in tests, nor does it obey the command line
2223
2433
override, so tests that accidentally write to the common directory should
2226
:cvar TEST_ROOT: Directory containing all temporary directories, plus
2227
a .bzr directory that stops us ascending higher into the filesystem.
2436
:cvar TEST_ROOT: Directory containing all temporary directories, plus a
2437
``.bzr`` directory that stops us ascending higher into the filesystem.
2230
2440
TEST_ROOT = None
2389
2600
root = TestCaseWithMemoryTransport.TEST_ROOT
2390
bzrdir.BzrDir.create_standalone_workingtree(root)
2601
# Make sure we get a readable and accessible home for .bzr.log
2602
# and/or config files, and not fallback to weird defaults (see
2603
# http://pad.lv/825027).
2604
self.assertIs(None, os.environ.get('BZR_HOME', None))
2605
os.environ['BZR_HOME'] = root
2606
wt = bzrdir.BzrDir.create_standalone_workingtree(root)
2607
del os.environ['BZR_HOME']
2608
# Hack for speed: remember the raw bytes of the dirstate file so that
2609
# we don't need to re-open the wt to check it hasn't changed.
2610
TestCaseWithMemoryTransport._SAFETY_NET_PRISTINE_DIRSTATE = (
2611
wt.control_transport.get_bytes('dirstate'))
2392
2613
def _check_safety_net(self):
2393
2614
"""Check that the safety .bzr directory have not been touched.
2440
2661
def make_branch(self, relpath, format=None):
2441
2662
"""Create a branch on the transport at relpath."""
2442
2663
repo = self.make_repository(relpath, format=format)
2443
return repo.bzrdir.create_branch()
2664
return repo.bzrdir.create_branch(append_revisions_only=False)
2666
def get_default_format(self):
2669
def resolve_format(self, format):
2670
"""Resolve an object to a ControlDir format object.
2672
The initial format object can either already be
2673
a ControlDirFormat, None (for the default format),
2674
or a string with the name of the control dir format.
2676
:param format: Object to resolve
2677
:return A ControlDirFormat instance
2680
format = self.get_default_format()
2681
if isinstance(format, basestring):
2682
format = bzrdir.format_registry.make_bzrdir(format)
2445
2685
def make_bzrdir(self, relpath, format=None):
2450
2690
t = _mod_transport.get_transport(maybe_a_url)
2451
2691
if len(segments) > 1 and segments[-1] not in ('', '.'):
2452
2692
t.ensure_base()
2455
if isinstance(format, basestring):
2456
format = bzrdir.format_registry.make_bzrdir(format)
2693
format = self.resolve_format(format)
2457
2694
return format.initialize_on_transport(t)
2458
2695
except errors.UninitializableFormat:
2459
2696
raise TestSkipped("Format %s is not initializable." % format)
2461
def make_repository(self, relpath, shared=False, format=None):
2698
def make_repository(self, relpath, shared=None, format=None):
2462
2699
"""Create a repository on our default transport at relpath.
2464
2701
Note that relpath must be a relative path, not a full url.
2498
2735
def setUp(self):
2499
2736
super(TestCaseWithMemoryTransport, self).setUp()
2500
# Ensure that ConnectedTransport doesn't leak sockets
2501
def get_transport_with_cleanup(*args, **kwargs):
2502
t = orig_get_transport(*args, **kwargs)
2503
if isinstance(t, _mod_transport.ConnectedTransport):
2504
self.addCleanup(t.disconnect)
2507
orig_get_transport = self.overrideAttr(_mod_transport, 'get_transport',
2508
get_transport_with_cleanup)
2738
def _add_disconnect_cleanup(transport):
2739
"""Schedule disconnection of given transport at test cleanup
2741
This needs to happen for all connected transports or leaks occur.
2743
Note reconnections may mean we call disconnect multiple times per
2744
transport which is suboptimal but seems harmless.
2746
self.addCleanup(transport.disconnect)
2748
_mod_transport.Transport.hooks.install_named_hook('post_connect',
2749
_add_disconnect_cleanup, None)
2509
2751
self._make_test_root()
2510
2752
self.addCleanup(os.chdir, os.getcwdu())
2511
2753
self.makeAndChdirToTestDir()
3139
3400
class TestDecorator(TestUtil.TestSuite):
3140
3401
"""A decorator for TestCase/TestSuite objects.
3142
Usually, subclasses should override __iter__(used when flattening test
3143
suites), which we do to filter, reorder, parallelise and so on, run() and
3403
Contains rather than flattening suite passed on construction
3147
def __init__(self, suite):
3148
TestUtil.TestSuite.__init__(self)
3151
def countTestCases(self):
3154
cases += test.countTestCases()
3161
def run(self, result):
3162
# Use iteration on self, not self._tests, to allow subclasses to hook
3165
if result.shouldStop:
3406
def __init__(self, suite=None):
3407
super(TestDecorator, self).__init__()
3408
if suite is not None:
3411
# Don't need subclass run method with suite emptying
3412
run = unittest.TestSuite.run
3171
3415
class CountingDecorator(TestDecorator):
3182
3426
"""A decorator which excludes test matching an exclude pattern."""
3184
3428
def __init__(self, suite, exclude_pattern):
3185
TestDecorator.__init__(self, suite)
3186
self.exclude_pattern = exclude_pattern
3187
self.excluded = False
3191
return iter(self._tests)
3192
self.excluded = True
3193
suite = exclude_tests_by_re(self, self.exclude_pattern)
3195
self.addTests(suite)
3196
return iter(self._tests)
3429
super(ExcludeDecorator, self).__init__(
3430
exclude_tests_by_re(suite, exclude_pattern))
3199
3433
class FilterTestsDecorator(TestDecorator):
3200
3434
"""A decorator which filters tests to those matching a pattern."""
3202
3436
def __init__(self, suite, pattern):
3203
TestDecorator.__init__(self, suite)
3204
self.pattern = pattern
3205
self.filtered = False
3209
return iter(self._tests)
3210
self.filtered = True
3211
suite = filter_suite_by_re(self, self.pattern)
3213
self.addTests(suite)
3214
return iter(self._tests)
3437
super(FilterTestsDecorator, self).__init__(
3438
filter_suite_by_re(suite, pattern))
3217
3441
class RandomDecorator(TestDecorator):
3218
3442
"""A decorator which randomises the order of its tests."""
3220
3444
def __init__(self, suite, random_seed, stream):
3221
TestDecorator.__init__(self, suite)
3222
self.random_seed = random_seed
3223
self.randomised = False
3224
self.stream = stream
3228
return iter(self._tests)
3229
self.randomised = True
3230
self.stream.write("Randomizing test order using seed %s\n\n" %
3231
(self.actual_seed()))
3445
random_seed = self.actual_seed(random_seed)
3446
stream.write("Randomizing test order using seed %s\n\n" %
3232
3448
# Initialise the random number generator.
3233
random.seed(self.actual_seed())
3234
suite = randomize_suite(self)
3236
self.addTests(suite)
3237
return iter(self._tests)
3449
random.seed(random_seed)
3450
super(RandomDecorator, self).__init__(randomize_suite(suite))
3239
def actual_seed(self):
3240
if self.random_seed == "now":
3453
def actual_seed(seed):
3241
3455
# We convert the seed to a long to make it reuseable across
3242
3456
# invocations (because the user can reenter it).
3243
self.random_seed = long(time.time())
3457
return long(time.time())
3245
3459
# Convert the seed to a long if we can
3247
self.random_seed = long(self.random_seed)
3462
except (TypeError, ValueError):
3250
return self.random_seed
3253
3467
class TestFirstDecorator(TestDecorator):
3254
3468
"""A decorator which moves named tests to the front."""
3256
3470
def __init__(self, suite, pattern):
3257
TestDecorator.__init__(self, suite)
3258
self.pattern = pattern
3259
self.filtered = False
3263
return iter(self._tests)
3264
self.filtered = True
3265
suites = split_suite_by_re(self, self.pattern)
3267
self.addTests(suites)
3268
return iter(self._tests)
3471
super(TestFirstDecorator, self).__init__()
3472
self.addTests(split_suite_by_re(suite, pattern))
3271
3475
def partition_tests(suite, count):
3316
3520
ProtocolTestCase.run(self, result)
3318
os.waitpid(self.pid, 0)
3522
pid, status = os.waitpid(self.pid, 0)
3523
# GZ 2011-10-18: If status is nonzero, should report to the result
3524
# that something went wrong.
3320
3526
test_blocks = partition_tests(suite, concurrency)
3527
# Clear the tests from the original suite so it doesn't keep them alive
3528
suite._tests[:] = []
3321
3529
for process_tests in test_blocks:
3322
process_suite = TestUtil.TestSuite()
3323
process_suite.addTests(process_tests)
3530
process_suite = TestUtil.TestSuite(process_tests)
3531
# Also clear each split list so new suite has only reference
3532
process_tests[:] = []
3324
3533
c2pread, c2pwrite = os.pipe()
3325
3534
pid = os.fork()
3327
workaround_zealous_crypto_random()
3537
stream = os.fdopen(c2pwrite, 'wb', 1)
3538
workaround_zealous_crypto_random()
3329
3539
os.close(c2pread)
3330
3540
# Leave stderr and stdout open so we can see test noise
3331
3541
# Close stdin so that the child goes away if it decides to
3332
3542
# read from stdin (otherwise its a roulette to see what
3333
3543
# child actually gets keystrokes for pdb etc).
3334
3544
sys.stdin.close()
3336
stream = os.fdopen(c2pwrite, 'wb', 1)
3337
3545
subunit_result = AutoTimingTestResultDecorator(
3338
TestProtocolClient(stream))
3546
SubUnitBzrProtocolClient(stream))
3339
3547
process_suite.run(subunit_result)
3549
# Try and report traceback on stream, but exit with error even
3550
# if stream couldn't be created or something else goes wrong.
3551
# The traceback is formatted to a string and written in one go
3552
# to avoid interleaving lines from multiple failing children.
3554
stream.write(traceback.format_exc())
3343
3559
os.close(c2pwrite)
3344
3560
stream = os.fdopen(c2pread, 'rb', 1)
3746
3966
'bzrlib.tests.test_commit_merge',
3747
3967
'bzrlib.tests.test_config',
3748
3968
'bzrlib.tests.test_conflicts',
3969
'bzrlib.tests.test_controldir',
3749
3970
'bzrlib.tests.test_counted_lock',
3750
3971
'bzrlib.tests.test_crash',
3751
3972
'bzrlib.tests.test_decorators',
3752
3973
'bzrlib.tests.test_delta',
3753
3974
'bzrlib.tests.test_debug',
3754
'bzrlib.tests.test_deprecated_graph',
3755
3975
'bzrlib.tests.test_diff',
3756
3976
'bzrlib.tests.test_directory_service',
3757
3977
'bzrlib.tests.test_dirstate',
3758
3978
'bzrlib.tests.test_email_message',
3759
3979
'bzrlib.tests.test_eol_filters',
3760
3980
'bzrlib.tests.test_errors',
3981
'bzrlib.tests.test_estimate_compressed_size',
3761
3982
'bzrlib.tests.test_export',
3983
'bzrlib.tests.test_export_pot',
3762
3984
'bzrlib.tests.test_extract',
3985
'bzrlib.tests.test_features',
3763
3986
'bzrlib.tests.test_fetch',
3764
3987
'bzrlib.tests.test_fixtures',
3765
3988
'bzrlib.tests.test_fifo_cache',
3766
3989
'bzrlib.tests.test_filters',
3990
'bzrlib.tests.test_filter_tree',
3767
3991
'bzrlib.tests.test_ftp_transport',
3768
3992
'bzrlib.tests.test_foreign',
3769
3993
'bzrlib.tests.test_generate_docs',
3875
4103
'bzrlib.tests.test_upgrade',
3876
4104
'bzrlib.tests.test_upgrade_stacked',
3877
4105
'bzrlib.tests.test_urlutils',
4106
'bzrlib.tests.test_utextwrap',
3878
4107
'bzrlib.tests.test_version',
3879
4108
'bzrlib.tests.test_version_info',
3880
4109
'bzrlib.tests.test_versionedfile',
4110
'bzrlib.tests.test_vf_search',
3881
4111
'bzrlib.tests.test_weave',
3882
4112
'bzrlib.tests.test_whitebox',
3883
4113
'bzrlib.tests.test_win32utils',
4205
4435
% (os.path.basename(dirname), printable_e))
4208
class Feature(object):
4209
"""An operating system Feature."""
4212
self._available = None
4214
def available(self):
4215
"""Is the feature available?
4217
:return: True if the feature is available.
4219
if self._available is None:
4220
self._available = self._probe()
4221
return self._available
4224
"""Implement this method in concrete features.
4226
:return: True if the feature is available.
4228
raise NotImplementedError
4231
if getattr(self, 'feature_name', None):
4232
return self.feature_name()
4233
return self.__class__.__name__
4236
class _SymlinkFeature(Feature):
4239
return osutils.has_symlinks()
4241
def feature_name(self):
4244
SymlinkFeature = _SymlinkFeature()
4247
class _HardlinkFeature(Feature):
4250
return osutils.has_hardlinks()
4252
def feature_name(self):
4255
HardlinkFeature = _HardlinkFeature()
4258
class _OsFifoFeature(Feature):
4261
return getattr(os, 'mkfifo', None)
4263
def feature_name(self):
4264
return 'filesystem fifos'
4266
OsFifoFeature = _OsFifoFeature()
4269
class _UnicodeFilenameFeature(Feature):
4270
"""Does the filesystem support Unicode filenames?"""
4274
# Check for character combinations unlikely to be covered by any
4275
# single non-unicode encoding. We use the characters
4276
# - greek small letter alpha (U+03B1) and
4277
# - braille pattern dots-123456 (U+283F).
4278
os.stat(u'\u03b1\u283f')
4279
except UnicodeEncodeError:
4281
except (IOError, OSError):
4282
# The filesystem allows the Unicode filename but the file doesn't
4286
# The filesystem allows the Unicode filename and the file exists,
4290
UnicodeFilenameFeature = _UnicodeFilenameFeature()
4293
class _CompatabilityThunkFeature(Feature):
4294
"""This feature is just a thunk to another feature.
4296
It issues a deprecation warning if it is accessed, to let you know that you
4297
should really use a different feature.
4300
def __init__(self, dep_version, module, name,
4301
replacement_name, replacement_module=None):
4302
super(_CompatabilityThunkFeature, self).__init__()
4303
self._module = module
4304
if replacement_module is None:
4305
replacement_module = module
4306
self._replacement_module = replacement_module
4308
self._replacement_name = replacement_name
4309
self._dep_version = dep_version
4310
self._feature = None
4313
if self._feature is None:
4314
depr_msg = self._dep_version % ('%s.%s'
4315
% (self._module, self._name))
4316
use_msg = ' Use %s.%s instead.' % (self._replacement_module,
4317
self._replacement_name)
4318
symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
4319
# Import the new feature and use it as a replacement for the
4321
self._feature = pyutils.get_named_object(
4322
self._replacement_module, self._replacement_name)
4326
return self._feature._probe()
4329
class ModuleAvailableFeature(Feature):
4330
"""This is a feature than describes a module we want to be available.
4332
Declare the name of the module in __init__(), and then after probing, the
4333
module will be available as 'self.module'.
4335
:ivar module: The module if it is available, else None.
4338
def __init__(self, module_name):
4339
super(ModuleAvailableFeature, self).__init__()
4340
self.module_name = module_name
4344
self._module = __import__(self.module_name, {}, {}, [''])
4351
if self.available(): # Make sure the probe has been done
4355
def feature_name(self):
4356
return self.module_name
4359
4438
def probe_unicode_in_user_encoding():
4360
4439
"""Try to encode several unicode strings to use in unicode-aware tests.
4361
4440
Return first successfull match.
4392
class _HTTPSServerFeature(Feature):
4393
"""Some tests want an https Server, check if one is available.
4395
Right now, the only way this is available is under python2.6 which provides
4406
def feature_name(self):
4407
return 'HTTPSServer'
4410
HTTPSServerFeature = _HTTPSServerFeature()
4413
class _UnicodeFilename(Feature):
4414
"""Does the filesystem support Unicode filenames?"""
4419
except UnicodeEncodeError:
4421
except (IOError, OSError):
4422
# The filesystem allows the Unicode filename but the file doesn't
4426
# The filesystem allows the Unicode filename and the file exists,
4430
UnicodeFilename = _UnicodeFilename()
4433
class _ByteStringNamedFilesystem(Feature):
4434
"""Is the filesystem based on bytes?"""
4437
if os.name == "posix":
4441
ByteStringNamedFilesystem = _ByteStringNamedFilesystem()
4444
class _UTF8Filesystem(Feature):
4445
"""Is the filesystem UTF-8?"""
4448
if osutils._fs_enc.upper() in ('UTF-8', 'UTF8'):
4452
UTF8Filesystem = _UTF8Filesystem()
4455
class _BreakinFeature(Feature):
4456
"""Does this platform support the breakin feature?"""
4459
from bzrlib import breakin
4460
if breakin.determine_signal() is None:
4462
if sys.platform == 'win32':
4463
# Windows doesn't have os.kill, and we catch the SIGBREAK signal.
4464
# We trigger SIGBREAK via a Console api so we need ctypes to
4465
# access the function
4472
def feature_name(self):
4473
return "SIGQUIT or SIGBREAK w/ctypes on win32"
4476
BreakinFeature = _BreakinFeature()
4479
class _CaseInsCasePresFilenameFeature(Feature):
4480
"""Is the file-system case insensitive, but case-preserving?"""
4483
fileno, name = tempfile.mkstemp(prefix='MixedCase')
4485
# first check truly case-preserving for created files, then check
4486
# case insensitive when opening existing files.
4487
name = osutils.normpath(name)
4488
base, rel = osutils.split(name)
4489
found_rel = osutils.canonical_relpath(base, name)
4490
return (found_rel == rel
4491
and os.path.isfile(name.upper())
4492
and os.path.isfile(name.lower()))
4497
def feature_name(self):
4498
return "case-insensitive case-preserving filesystem"
4500
CaseInsCasePresFilenameFeature = _CaseInsCasePresFilenameFeature()
4503
class _CaseInsensitiveFilesystemFeature(Feature):
4504
"""Check if underlying filesystem is case-insensitive but *not* case
4507
# Note that on Windows, Cygwin, MacOS etc, the file-systems are far
4508
# more likely to be case preserving, so this case is rare.
4511
if CaseInsCasePresFilenameFeature.available():
4514
if TestCaseWithMemoryTransport.TEST_ROOT is None:
4515
root = osutils.mkdtemp(prefix='testbzr-', suffix='.tmp')
4516
TestCaseWithMemoryTransport.TEST_ROOT = root
4518
root = TestCaseWithMemoryTransport.TEST_ROOT
4519
tdir = osutils.mkdtemp(prefix='case-sensitive-probe-', suffix='',
4521
name_a = osutils.pathjoin(tdir, 'a')
4522
name_A = osutils.pathjoin(tdir, 'A')
4524
result = osutils.isdir(name_A)
4525
_rmtree_temp_dir(tdir)
4528
def feature_name(self):
4529
return 'case-insensitive filesystem'
4531
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
4534
class _CaseSensitiveFilesystemFeature(Feature):
4537
if CaseInsCasePresFilenameFeature.available():
4539
elif CaseInsensitiveFilesystemFeature.available():
4544
def feature_name(self):
4545
return 'case-sensitive filesystem'
4547
# new coding style is for feature instances to be lowercase
4548
case_sensitive_filesystem_feature = _CaseSensitiveFilesystemFeature()
4551
4471
# Only define SubUnitBzrRunner if subunit is available.
4553
4473
from subunit import TestProtocolClient
4554
4474
from subunit.test_results import AutoTimingTestResultDecorator
4555
4475
class SubUnitBzrProtocolClient(TestProtocolClient):
4477
def stopTest(self, test):
4478
super(SubUnitBzrProtocolClient, self).stopTest(test)
4479
_clear__type_equality_funcs(test)
4557
4481
def addSuccess(self, test, details=None):
4558
4482
# The subunit client always includes the details in the subunit
4559
4483
# stream, but we don't want to include it in ours.
4571
4495
except ImportError:
4574
class _PosixPermissionsFeature(Feature):
4578
# create temporary file and check if specified perms are maintained.
4581
write_perms = stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR
4582
f = tempfile.mkstemp(prefix='bzr_perms_chk_')
4585
os.chmod(name, write_perms)
4587
read_perms = os.stat(name).st_mode & 0777
4589
return (write_perms == read_perms)
4591
return (os.name == 'posix') and has_perms()
4593
def feature_name(self):
4594
return 'POSIX permissions support'
4596
posix_permissions_feature = _PosixPermissionsFeature()
4499
# API compatibility for old plugins; see bug 892622.
4502
'HTTPServerFeature',
4503
'ModuleAvailableFeature',
4504
'HTTPSServerFeature', 'SymlinkFeature', 'HardlinkFeature',
4505
'OsFifoFeature', 'UnicodeFilenameFeature',
4506
'ByteStringNamedFilesystem', 'UTF8Filesystem',
4507
'BreakinFeature', 'CaseInsCasePresFilenameFeature',
4508
'CaseInsensitiveFilesystemFeature', 'case_sensitive_filesystem_feature',
4509
'posix_permissions_feature',
4511
globals()[name] = _CompatabilityThunkFeature(
4512
symbol_versioning.deprecated_in((2, 5, 0)),
4513
'bzrlib.tests', name,
4514
name, 'bzrlib.tests.features')
4517
for (old_name, new_name) in [
4518
('UnicodeFilename', 'UnicodeFilenameFeature'),
4520
globals()[name] = _CompatabilityThunkFeature(
4521
symbol_versioning.deprecated_in((2, 5, 0)),
4522
'bzrlib.tests', old_name,
4523
new_name, 'bzrlib.tests.features')
added
removed
bzrlib/tests/__init__.py
