117
from bzrlib.tests.http_server import HttpServer
118
from bzrlib.tests.TestUtil import (
122
117
from bzrlib.ui import NullProgressView
123
118
from bzrlib.ui.text import TextUIFactory
124
119
import bzrlib.version_info_formats.format_custom
140
135
SUBUNIT_SEEK_SET = 0
141
136
SUBUNIT_SEEK_CUR = 1
138
# These are intentionally brought into this namespace. That way plugins, etc
139
# can just "from bzrlib.tests import TestCase, TestLoader, etc"
140
TestSuite = TestUtil.TestSuite
141
TestLoader = TestUtil.TestLoader
144
143
class ExtendedTestResult(testtools.TextTestResult):
145
144
"""Accepts, reports and accumulates the results of running tests.
848
847
# going away but leak one) but it seems less likely than the actual
849
848
# false positives (the test see threads going away and does not leak).
850
849
if leaked_threads > 0:
850
if 'threads' in selftest_debug_flags:
851
print '%s is leaking, active is now %d' % (self.id(), active)
851
852
TestCase._leaking_threads_tests += 1
852
853
if TestCase._first_thread_leaker_id is None:
853
854
TestCase._first_thread_leaker_id = self.id()
1458
1459
The file is removed as the test is torn down.
1460
fileno, name = tempfile.mkstemp(suffix='.log', prefix='testbzr')
1461
self._log_file = os.fdopen(fileno, 'w+')
1461
self._log_file = StringIO()
1462
1462
self._log_memento = bzrlib.trace.push_log_file(self._log_file)
1463
self._log_file_name = name
1464
1463
self.addCleanup(self._finishLogFile)
1466
1465
def _finishLogFile(self):
1668
1667
unicodestr = self._log_contents.decode('utf8', 'replace')
1669
1668
self._log_contents = unicodestr.encode('utf8')
1670
1669
return self._log_contents
1672
if bzrlib.trace._trace_file:
1673
# flush the log file, to get all content
1674
bzrlib.trace._trace_file.flush()
1675
if self._log_file_name is not None:
1676
logfile = open(self._log_file_name)
1678
log_contents = logfile.read()
1670
if self._log_file is not None:
1671
log_contents = self._log_file.getvalue()
1682
1673
log_contents.decode('utf8')
1683
1674
except UnicodeDecodeError:
1684
1675
unicodestr = log_contents.decode('utf8', 'replace')
1685
1676
log_contents = unicodestr.encode('utf8')
1686
1677
if not keep_log_file:
1688
max_close_attempts = 100
1689
first_close_error = None
1690
while close_attempts < max_close_attempts:
1693
self._log_file.close()
1694
except IOError, ioe:
1695
if ioe.errno is None:
1696
# No errno implies 'close() called during
1697
# concurrent operation on the same file object', so
1698
# retry. Probably a thread is trying to write to
1700
if first_close_error is None:
1701
first_close_error = ioe
1706
if close_attempts > 1:
1708
'Unable to close log file on first attempt, '
1709
'will retry: %s\n' % (first_close_error,))
1710
if close_attempts == max_close_attempts:
1712
'Unable to close log file after %d attempts.\n'
1713
% (max_close_attempts,))
1714
1678
self._log_file = None
1715
1679
# Permit multiple calls to get_log until we clean it up in
1716
1680
# finishLogFile
1717
1681
self._log_contents = log_contents
1719
os.remove(self._log_file_name)
1721
if sys.platform == 'win32' and e.errno == errno.EACCES:
1722
sys.stderr.write(('Unable to delete log file '
1723
' %r\n' % self._log_file_name))
1726
self._log_file_name = None
1727
1682
return log_contents
1729
return "No log file content and no log file name."
1684
return "No log file content."
1731
1686
def get_log(self):
1732
1687
"""Get a unicode string containing the log from bzrlib.trace.
1947
1902
variables. A value of None will unset the env variable.
1948
1903
The values must be strings. The change will only occur in the
1949
1904
child, so you don't need to fix the environment after running.
1950
:param skip_if_plan_to_signal: raise TestSkipped when true and os.kill
1905
:param skip_if_plan_to_signal: raise TestSkipped when true and system
1906
doesn't support signalling subprocesses.
1952
1907
:param allow_plugins: If False (default) pass --no-plugins to bzr.
1954
1909
:returns: Popen object for the started process.
1956
1911
if skip_if_plan_to_signal:
1957
if not getattr(os, 'kill', None):
1958
raise TestSkipped("os.kill not available.")
1912
if os.name != "posix":
1913
raise TestSkipped("Sending signals not supported")
1960
1915
if env_changes is None:
1961
1916
env_changes = {}
2440
2395
def setUp(self):
2441
2396
super(TestCaseWithMemoryTransport, self).setUp()
2397
# Ensure that ConnectedTransport doesn't leak sockets
2398
def get_transport_with_cleanup(*args, **kwargs):
2399
t = orig_get_transport(*args, **kwargs)
2400
if isinstance(t, _mod_transport.ConnectedTransport):
2401
self.addCleanup(t.disconnect)
2404
orig_get_transport = self.overrideAttr(_mod_transport, 'get_transport',
2405
get_transport_with_cleanup)
2442
2406
self._make_test_root()
2443
2407
self.addCleanup(os.chdir, os.getcwdu())
2444
2408
self.makeAndChdirToTestDir()
2738
2702
def setUp(self):
2703
from bzrlib.tests import http_server
2739
2704
super(ChrootedTestCase, self).setUp()
2740
2705
if not self.vfs_transport_factory == memory.MemoryServer:
2741
self.transport_readonly_server = HttpServer
2706
self.transport_readonly_server = http_server.HttpServer
2744
2709
def condition_id_re(pattern):
3068
class TestDecorator(TestSuite):
3033
class TestDecorator(TestUtil.TestSuite):
3069
3034
"""A decorator for TestCase/TestSuite objects.
3071
3036
Usually, subclasses should override __iter__(used when flattening test
3249
3214
test_blocks = partition_tests(suite, concurrency)
3250
3215
for process_tests in test_blocks:
3251
process_suite = TestSuite()
3216
process_suite = TestUtil.TestSuite()
3252
3217
process_suite.addTests(process_tests)
3253
3218
c2pread, c2pwrite = os.pipe()
3254
3219
pid = os.fork()
3409
3374
# rather than failing tests. And no longer raise
3410
3375
# LockContention when fctnl locks are not being used
3411
3376
# with proper exclusion rules.
3377
# -Ethreads Will display thread ident at creation/join time to
3378
# help track thread leaks
3412
3379
selftest_debug_flags = set()
3809
3776
'bzrlib.tests.test_switch',
3810
3777
'bzrlib.tests.test_symbol_versioning',
3811
3778
'bzrlib.tests.test_tag',
3779
'bzrlib.tests.test_test_server',
3812
3780
'bzrlib.tests.test_testament',
3813
3781
'bzrlib.tests.test_textfile',
3814
3782
'bzrlib.tests.test_textmerge',
4005
3973
... bzrlib.tests.test_sampler.DemoTest('test_nothing'),
4006
3974
... [('one', dict(param=1)),
4007
3975
... ('two', dict(param=2))],
3976
... TestUtil.TestSuite())
4009
3977
>>> tests = list(iter_suite_tests(r))
added
removed
bzrlib/tests/__init__.py
