1
# Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Canonical Ltd
1
# Copyright (C) 2005-2010 Canonical Ltd
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
101
101
deprecated_passed,
103
103
import bzrlib.trace
104
from bzrlib.transport import get_transport, pathfilter
104
from bzrlib.transport import (
105
109
import bzrlib.transport
106
from bzrlib.transport.local import LocalURLServer
107
from bzrlib.transport.memory import MemoryServer
108
from bzrlib.transport.readonly import ReadonlyServer
109
110
from bzrlib.trace import mutter, note
110
from bzrlib.tests import TestUtil
111
from bzrlib.tests import (
111
115
from bzrlib.tests.http_server import HttpServer
112
116
from bzrlib.tests.TestUtil import (
124
128
# shown frame is the test code, not our assertXYZ.
127
default_transport = LocalURLServer
131
default_transport = test_server.LocalURLServer
130
134
_unitialized_attr = object()
485
489
return self._shortened_test_description(test)
487
491
def report_error(self, test, err):
488
ui.ui_factory.note('ERROR: %s\n %s\n' % (
492
self.ui.note('ERROR: %s\n %s\n' % (
489
493
self._test_description(test),
493
497
def report_failure(self, test, err):
494
ui.ui_factory.note('FAIL: %s\n %s\n' % (
498
self.ui.note('FAIL: %s\n %s\n' % (
495
499
self._test_description(test),
1044
1044
if t.base.endswith('/work/'):
1045
1045
# we have safety net/test root/work
1046
1046
t = t.clone('../..')
1047
elif isinstance(transport_server, server.SmartTCPServer_for_testing):
1047
elif isinstance(transport_server,
1048
test_server.SmartTCPServer_for_testing):
1048
1049
# The smart server adds a path similar to work, which is traversed
1049
1050
# up from by the client. But the server is chrooted - the actual
1050
1051
# backing transport is not escaped from, and VFS requests to the
1205
1206
raise AssertionError('pattern "%s" found in "%s"'
1206
1207
% (needle_re, haystack))
1209
def assertContainsString(self, haystack, needle):
1210
if haystack.find(needle) == -1:
1211
self.fail("string %r not found in '''%s'''" % (needle, haystack))
1208
1213
def assertSubset(self, sublist, superlist):
1209
1214
"""Assert that every entry in sublist is present in superlist."""
1210
1215
missing = set(sublist) - set(superlist)
1540
1545
'ftp_proxy': None,
1541
1546
'FTP_PROXY': None,
1542
1547
'BZR_REMOTE_PATH': None,
1548
# Generally speaking, we don't want apport reporting on crashes in
1549
# the test envirnoment unless we're specifically testing apport,
1550
# so that it doesn't leak into the real system environment. We
1551
# use an env var so it propagates to subprocesses.
1552
'APPORT_DISABLE': '1',
1545
1555
self.addCleanup(self._restoreEnvironment)
1546
1556
for name, value in new_env.iteritems():
1547
1557
self._captureVar(name, value)
1549
1559
def _captureVar(self, name, newvalue):
1550
1560
"""Set an environment variable, and reset it when finished."""
1551
self.__old_env[name] = osutils.set_or_unset_env(name, newvalue)
1561
self._old_env[name] = osutils.set_or_unset_env(name, newvalue)
1553
1563
def _restoreEnvironment(self):
1554
for name, value in self.__old_env.iteritems():
1564
for name, value in self._old_env.iteritems():
1555
1565
osutils.set_or_unset_env(name, value)
1557
1567
def _restoreHooks(self):
2170
2180
if self.__readonly_server is None:
2171
2181
if self.transport_readonly_server is None:
2172
2182
# readonly decorator requested
2173
self.__readonly_server = ReadonlyServer()
2183
self.__readonly_server = test_server.ReadonlyServer()
2175
2185
# explicit readonly transport.
2176
2186
self.__readonly_server = self.create_transport_readonly_server()
2199
2209
is no means to override it.
2201
2211
if self.__vfs_server is None:
2202
self.__vfs_server = MemoryServer()
2212
self.__vfs_server = memory.MemoryServer()
2203
2213
self.start_server(self.__vfs_server)
2204
2214
return self.__vfs_server
2362
2372
return made_control.create_repository(shared=shared)
2364
2374
def make_smart_server(self, path):
2365
smart_server = server.SmartTCPServer_for_testing()
2375
smart_server = test_server.SmartTCPServer_for_testing()
2366
2376
self.start_server(smart_server, self.get_server())
2367
2377
remote_transport = get_transport(smart_server.get_url()).clone(path)
2368
2378
return remote_transport
2396
2406
def setup_smart_server_with_call_log(self):
2397
2407
"""Sets up a smart server as the transport server with a call log."""
2398
self.transport_server = server.SmartTCPServer_for_testing
2408
self.transport_server = test_server.SmartTCPServer_for_testing
2399
2409
self.hpss_calls = []
2400
2410
import traceback
2401
2411
# Skip the current stack down to the caller of
2614
2624
# We can only make working trees locally at the moment. If the
2615
2625
# transport can't support them, then we keep the non-disk-backed
2616
2626
# branch and create a local checkout.
2617
if self.vfs_transport_factory is LocalURLServer:
2627
if self.vfs_transport_factory is test_server.LocalURLServer:
2618
2628
# the branch is colocated on disk, we cannot create a checkout.
2619
2629
# hopefully callers will expect this.
2620
2630
local_controldir= bzrdir.BzrDir.open(self.get_vfs_only_url(relpath))
2681
2691
def setUp(self):
2682
2692
super(ChrootedTestCase, self).setUp()
2683
if not self.vfs_transport_factory == MemoryServer:
2693
if not self.vfs_transport_factory == memory.MemoryServer:
2684
2694
self.transport_readonly_server = HttpServer
3557
3567
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
3558
3568
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
3560
# Obvious higest levels prefixes, feel free to add your own via a plugin
3570
# Obvious highest levels prefixes, feel free to add your own via a plugin
3561
3571
test_prefix_alias_registry.register('bd', 'bzrlib.doc')
3562
3572
test_prefix_alias_registry.register('bu', 'bzrlib.utils')
3563
3573
test_prefix_alias_registry.register('bt', 'bzrlib.tests')
3657
3667
'bzrlib.tests.test_identitymap',
3658
3668
'bzrlib.tests.test_ignores',
3659
3669
'bzrlib.tests.test_index',
3670
'bzrlib.tests.test_import_tariff',
3660
3671
'bzrlib.tests.test_info',
3661
3672
'bzrlib.tests.test_inv',
3662
3673
'bzrlib.tests.test_inventory_delta',
4130
4141
should really use a different feature.
4133
def __init__(self, module, name, this_name, dep_version):
4144
def __init__(self, dep_version, module, name,
4145
replacement_name, replacement_module=None):
4134
4146
super(_CompatabilityThunkFeature, self).__init__()
4135
4147
self._module = module
4148
if replacement_module is None:
4149
replacement_module = module
4150
self._replacement_module = replacement_module
4136
4151
self._name = name
4137
self._this_name = this_name
4152
self._replacement_name = replacement_name
4138
4153
self._dep_version = dep_version
4139
4154
self._feature = None
4141
4156
def _ensure(self):
4142
4157
if self._feature is None:
4143
msg = (self._dep_version % self._this_name) + (
4144
' Use %s.%s instead.' % (self._module, self._name))
4145
symbol_versioning.warn(msg, DeprecationWarning)
4146
mod = __import__(self._module, {}, {}, [self._name])
4147
self._feature = getattr(mod, self._name)
4158
depr_msg = self._dep_version % ('%s.%s'
4159
% (self._module, self._name))
4160
use_msg = ' Use %s.%s instead.' % (self._replacement_module,
4161
self._replacement_name)
4162
symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
4163
# Import the new feature and use it as a replacement for the
4165
mod = __import__(self._replacement_module, {}, {},
4166
[self._replacement_name])
4167
self._feature = getattr(mod, self._replacement_name)
4149
4169
def _probe(self):
4176
4196
if self.available(): # Make sure the probe has been done
4177
4197
return self._module
4180
4200
def feature_name(self):
4181
4201
return self.module_name
4184
4204
# This is kept here for compatibility, it is recommended to use
4185
4205
# 'bzrlib.tests.feature.paramiko' instead
4186
ParamikoFeature = _CompatabilityThunkFeature('bzrlib.tests.features',
4187
'paramiko', 'bzrlib.tests.ParamikoFeature', deprecated_in((2,1,0)))
4206
ParamikoFeature = _CompatabilityThunkFeature(
4207
deprecated_in((2,1,0)),
4208
'bzrlib.tests.features', 'ParamikoFeature', 'paramiko')
4190
4211
def probe_unicode_in_user_encoding():
4351
4372
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
4375
class _CaseSensitiveFilesystemFeature(Feature):
4378
if CaseInsCasePresFilenameFeature.available():
4380
elif CaseInsensitiveFilesystemFeature.available():
4385
def feature_name(self):
4386
return 'case-sensitive filesystem'
4388
# new coding style is for feature instances to be lowercase
4389
case_sensitive_filesystem_feature = _CaseSensitiveFilesystemFeature()
4354
4392
# Kept for compatibility, use bzrlib.tests.features.subunit instead
4355
SubUnitFeature = _CompatabilityThunkFeature('bzrlib.tests.features', 'subunit',
4356
'bzrlib.tests.SubUnitFeature', deprecated_in((2,1,0)))
4393
SubUnitFeature = _CompatabilityThunkFeature(
4394
deprecated_in((2,1,0)),
4395
'bzrlib.tests.features', 'SubUnitFeature', 'subunit')
4357
4396
# Only define SubUnitBzrRunner if subunit is available.
4359
4398
from subunit import TestProtocolClient
added
removed
bzrlib/tests/__init__.py
