~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/__init__.py

  • Committer: Martin Pool
  • Date: 2010-02-23 07:43:11 UTC
  • mfrom: (4797.2.20 2.1)
  • mto: This revision was merged to the branch mainline in revision 5055.
  • Revision ID: mbp@sourcefrog.net-20100223074311-gnj55xdhrgz9l94e
Merge 2.1 back to trunk

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2005, 2006, 2007, 2008, 2009, 2010 Canonical Ltd
 
1
# Copyright (C) 2005-2010 Canonical Ltd
2
2
#
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,
102
102
    )
103
103
import bzrlib.trace
104
 
from bzrlib.transport import get_transport, pathfilter
 
104
from bzrlib.transport import (
 
105
    get_transport,
 
106
    memory,
 
107
    pathfilter,
 
108
    )
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 (
 
112
    test_server,
 
113
    TestUtil,
 
114
    )
111
115
from bzrlib.tests.http_server import HttpServer
112
116
from bzrlib.tests.TestUtil import (
113
117
                          TestSuite,
124
128
# shown frame is the test code, not our assertXYZ.
125
129
__unittest = 1
126
130
 
127
 
default_transport = LocalURLServer
 
131
default_transport = test_server.LocalURLServer
128
132
 
129
133
 
130
134
_unitialized_attr = object()
485
489
        return self._shortened_test_description(test)
486
490
 
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),
490
494
            err[1],
491
495
            ))
492
496
 
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),
496
500
            err[1],
497
501
            ))
702
706
    """
703
707
 
704
708
 
705
 
class CommandFailed(Exception):
706
 
    pass
707
 
 
708
 
 
709
709
class StringIOWrapper(object):
710
710
    """A wrapper around cStringIO which just adds an encoding attribute.
711
711
 
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))
1207
1208
 
 
1209
    def assertContainsString(self, haystack, needle):
 
1210
        if haystack.find(needle) == -1:
 
1211
            self.fail("string %r not found in '''%s'''" % (needle, haystack))
 
1212
 
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',
1543
1553
        }
1544
 
        self.__old_env = {}
 
1554
        self._old_env = {}
1545
1555
        self.addCleanup(self._restoreEnvironment)
1546
1556
        for name, value in new_env.iteritems():
1547
1557
            self._captureVar(name, value)
1548
1558
 
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)
1552
1562
 
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)
1556
1566
 
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()
2174
2184
            else:
2175
2185
                # explicit readonly transport.
2176
2186
                self.__readonly_server = self.create_transport_readonly_server()
2199
2209
        is no means to override it.
2200
2210
        """
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
2205
2215
 
2362
2372
        return made_control.create_repository(shared=shared)
2363
2373
 
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
2395
2405
 
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))
2680
2690
 
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
2685
2695
 
2686
2696
 
3557
3567
# appear prefixed ('bzrlib.' is "replaced" by 'bzrlib.').
3558
3568
test_prefix_alias_registry.register('bzrlib', 'bzrlib')
3559
3569
 
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')
3618
3628
        'bzrlib.tests.test_chunk_writer',
3619
3629
        'bzrlib.tests.test_clean_tree',
3620
3630
        'bzrlib.tests.test_cleanup',
 
3631
        'bzrlib.tests.test_cmdline',
3621
3632
        'bzrlib.tests.test_commands',
3622
3633
        'bzrlib.tests.test_commit',
3623
3634
        'bzrlib.tests.test_commit_merge',
3657
3668
        'bzrlib.tests.test_identitymap',
3658
3669
        'bzrlib.tests.test_ignores',
3659
3670
        'bzrlib.tests.test_index',
 
3671
        'bzrlib.tests.test_import_tariff',
3660
3672
        'bzrlib.tests.test_info',
3661
3673
        'bzrlib.tests.test_inv',
3662
3674
        'bzrlib.tests.test_inventory_delta',
4130
4142
    should really use a different feature.
4131
4143
    """
4132
4144
 
4133
 
    def __init__(self, module, name, this_name, dep_version):
 
4145
    def __init__(self, dep_version, module, name,
 
4146
                 replacement_name, replacement_module=None):
4134
4147
        super(_CompatabilityThunkFeature, self).__init__()
4135
4148
        self._module = module
 
4149
        if replacement_module is None:
 
4150
            replacement_module = module
 
4151
        self._replacement_module = replacement_module
4136
4152
        self._name = name
4137
 
        self._this_name = this_name
 
4153
        self._replacement_name = replacement_name
4138
4154
        self._dep_version = dep_version
4139
4155
        self._feature = None
4140
4156
 
4141
4157
    def _ensure(self):
4142
4158
        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)
 
4159
            depr_msg = self._dep_version % ('%s.%s'
 
4160
                                            % (self._module, self._name))
 
4161
            use_msg = ' Use %s.%s instead.' % (self._replacement_module,
 
4162
                                               self._replacement_name)
 
4163
            symbol_versioning.warn(depr_msg + use_msg, DeprecationWarning)
 
4164
            # Import the new feature and use it as a replacement for the
 
4165
            # deprecated one.
 
4166
            mod = __import__(self._replacement_module, {}, {},
 
4167
                             [self._replacement_name])
 
4168
            self._feature = getattr(mod, self._replacement_name)
4148
4169
 
4149
4170
    def _probe(self):
4150
4171
        self._ensure()
4176
4197
        if self.available(): # Make sure the probe has been done
4177
4198
            return self._module
4178
4199
        return None
4179
 
    
 
4200
 
4180
4201
    def feature_name(self):
4181
4202
        return self.module_name
4182
4203
 
4183
4204
 
4184
4205
# This is kept here for compatibility, it is recommended to use
4185
4206
# 'bzrlib.tests.feature.paramiko' instead
4186
 
ParamikoFeature = _CompatabilityThunkFeature('bzrlib.tests.features',
4187
 
    'paramiko', 'bzrlib.tests.ParamikoFeature', deprecated_in((2,1,0)))
 
4207
ParamikoFeature = _CompatabilityThunkFeature(
 
4208
    deprecated_in((2,1,0)),
 
4209
    'bzrlib.tests.features', 'ParamikoFeature', 'paramiko')
4188
4210
 
4189
4211
 
4190
4212
def probe_unicode_in_user_encoding():
4351
4373
CaseInsensitiveFilesystemFeature = _CaseInsensitiveFilesystemFeature()
4352
4374
 
4353
4375
 
 
4376
class _CaseSensitiveFilesystemFeature(Feature):
 
4377
 
 
4378
    def _probe(self):
 
4379
        if CaseInsCasePresFilenameFeature.available():
 
4380
            return False
 
4381
        elif CaseInsensitiveFilesystemFeature.available():
 
4382
            return False
 
4383
        else:
 
4384
            return True
 
4385
 
 
4386
    def feature_name(self):
 
4387
        return 'case-sensitive filesystem'
 
4388
 
 
4389
# new coding style is for feature instances to be lowercase
 
4390
case_sensitive_filesystem_feature = _CaseSensitiveFilesystemFeature()
 
4391
 
 
4392
 
4354
4393
# 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)))
 
4394
SubUnitFeature = _CompatabilityThunkFeature(
 
4395
    deprecated_in((2,1,0)),
 
4396
    'bzrlib.tests.features', 'SubUnitFeature', 'subunit')
4357
4397
# Only define SubUnitBzrRunner if subunit is available.
4358
4398
try:
4359
4399
    from subunit import TestProtocolClient