← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/test_server.py

  • Committer: Vincent Ladeuil
  • Date: 2010-04-12 16:41:03 UTC
  • mto: (5148.1.1 integration)
  • mto: This revision was merged to the branch mainline in revision 5151.
  • Revision ID: v.ladeuil+lp@free.fr-20100412164103-v157103xtwozjf7n
Failing tests for bug #519319.

* bzrlib/tests/blackbox/test_send.py:
(TestSendStrictMixin.assertSendSucceeds): We want to be able to
succeed with a warning.
(TestSendStrictWithChanges.test_send_default)
(TestSendStrictWithChanges.test_send_bogus_config_var_ignored):
Succeed with a warning.

* bzrlib/tests/blackbox/test_push.py:
(TestPushStrictMixin.assertPushSucceeds): We want to be able to
succeed with a warning.
(TestPushStrictWithChanges.test_push_default): By default we
succeed with a warning.
(TestPushStrictWithChanges.test_push_bogus_config_var_ignored):
Ignoring a conf variable also succeeds with a warning.

* bzrlib/tests/blackbox/test_dpush.py:
(TestDpushStrictMixin.assertPushSucceeds): We want to be able to
succeed with a warning.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/test_server.py
 
1
# Copyright (C) 2005, 2006, 2007, 2008, 2010 Canonical Ltd
 
2
#
 
3
# This program is free software; you can redistribute it and/or modify
 
4
# it under the terms of the GNU General Public License as published by
 
5
# the Free Software Foundation; either version 2 of the License, or
 
6
# (at your option) any later version.
 
7
#
 
8
# This program is distributed in the hope that it will be useful,
 
9
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
# GNU General Public License for more details.
 
12
#
 
13
# You should have received a copy of the GNU General Public License
 
14
# along with this program; if not, write to the Free Software
 
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
 
16
 
 
17
from bzrlib import (
 
18
    transport,
 
19
    urlutils,
 
20
    )
 
21
from bzrlib.transport import (
 
22
    chroot,
 
23
    pathfilter,
 
24
    )
 
25
from bzrlib.smart import server
 
26
 
 
27
 
 
28
class TestServer(transport.Server):
 
29
    """A Transport Server dedicated to tests.
 
30
 
 
31
    The TestServer interface provides a server for a given transport. We use
 
32
    these servers as loopback testing tools. For any given transport the
 
33
    Servers it provides must either allow writing, or serve the contents
 
34
    of os.getcwdu() at the time start_server is called.
 
35
 
 
36
    Note that these are real servers - they must implement all the things
 
37
    that we want bzr transports to take advantage of.
 
38
    """
 
39
 
 
40
    def get_url(self):
 
41
        """Return a url for this server.
 
42
 
 
43
        If the transport does not represent a disk directory (i.e. it is
 
44
        a database like svn, or a memory only transport, it should return
 
45
        a connection to a newly established resource for this Server.
 
46
        Otherwise it should return a url that will provide access to the path
 
47
        that was os.getcwdu() when start_server() was called.
 
48
 
 
49
        Subsequent calls will return the same resource.
 
50
        """
 
51
        raise NotImplementedError
 
52
 
 
53
    def get_bogus_url(self):
 
54
        """Return a url for this protocol, that will fail to connect.
 
55
 
 
56
        This may raise NotImplementedError to indicate that this server cannot
 
57
        provide bogus urls.
 
58
        """
 
59
        raise NotImplementedError
 
60
 
 
61
 
 
62
class LocalURLServer(TestServer):
 
63
    """A pretend server for local transports, using file:// urls.
 
64
 
 
65
    Of course no actual server is required to access the local filesystem, so
 
66
    this just exists to tell the test code how to get to it.
 
67
    """
 
68
 
 
69
    def start_server(self):
 
70
        pass
 
71
 
 
72
    def get_url(self):
 
73
        """See Transport.Server.get_url."""
 
74
        return urlutils.local_path_to_url('')
 
75
 
 
76
 
 
77
class DecoratorServer(TestServer):
 
78
    """Server for the TransportDecorator for testing with.
 
79
 
 
80
    To use this when subclassing TransportDecorator, override override the
 
81
    get_decorator_class method.
 
82
    """
 
83
 
 
84
    def start_server(self, server=None):
 
85
        """See bzrlib.transport.Server.start_server.
 
86
 
 
87
        :server: decorate the urls given by server. If not provided a
 
88
        LocalServer is created.
 
89
        """
 
90
        if server is not None:
 
91
            self._made_server = False
 
92
            self._server = server
 
93
        else:
 
94
            self._made_server = True
 
95
            self._server = LocalURLServer()
 
96
            self._server.start_server()
 
97
 
 
98
    def stop_server(self):
 
99
        if self._made_server:
 
100
            self._server.stop_server()
 
101
 
 
102
    def get_decorator_class(self):
 
103
        """Return the class of the decorators we should be constructing."""
 
104
        raise NotImplementedError(self.get_decorator_class)
 
105
 
 
106
    def get_url_prefix(self):
 
107
        """What URL prefix does this decorator produce?"""
 
108
        return self.get_decorator_class()._get_url_prefix()
 
109
 
 
110
    def get_bogus_url(self):
 
111
        """See bzrlib.transport.Server.get_bogus_url."""
 
112
        return self.get_url_prefix() + self._server.get_bogus_url()
 
113
 
 
114
    def get_url(self):
 
115
        """See bzrlib.transport.Server.get_url."""
 
116
        return self.get_url_prefix() + self._server.get_url()
 
117
 
 
118
 
 
119
class BrokenRenameServer(DecoratorServer):
 
120
    """Server for the BrokenRenameTransportDecorator for testing with."""
 
121
 
 
122
    def get_decorator_class(self):
 
123
        from bzrlib.transport import brokenrename
 
124
        return brokenrename.BrokenRenameTransportDecorator
 
125
 
 
126
 
 
127
class FakeNFSServer(DecoratorServer):
 
128
    """Server for the FakeNFSTransportDecorator for testing with."""
 
129
 
 
130
    def get_decorator_class(self):
 
131
        from bzrlib.transport import fakenfs
 
132
        return fakenfs.FakeNFSTransportDecorator
 
133
 
 
134
 
 
135
class FakeVFATServer(DecoratorServer):
 
136
    """A server that suggests connections through FakeVFATTransportDecorator
 
137
 
 
138
    For use in testing.
 
139
    """
 
140
 
 
141
    def get_decorator_class(self):
 
142
        from bzrlib.transport import fakevfat
 
143
        return fakevfat.FakeVFATTransportDecorator
 
144
 
 
145
 
 
146
class LogDecoratorServer(DecoratorServer):
 
147
    """Server for testing."""
 
148
 
 
149
    def get_decorator_class(self):
 
150
        from bzrlib.transport import log
 
151
        return log.TransportLogDecorator
 
152
 
 
153
 
 
154
class NoSmartTransportServer(DecoratorServer):
 
155
    """Server for the NoSmartTransportDecorator for testing with."""
 
156
 
 
157
    def get_decorator_class(self):
 
158
        from bzrlib.transport import nosmart
 
159
        return nosmart.NoSmartTransportDecorator
 
160
 
 
161
 
 
162
class ReadonlyServer(DecoratorServer):
 
163
    """Server for the ReadonlyTransportDecorator for testing with."""
 
164
 
 
165
    def get_decorator_class(self):
 
166
        from bzrlib.transport import readonly
 
167
        return readonly.ReadonlyTransportDecorator
 
168
 
 
169
 
 
170
class TraceServer(DecoratorServer):
 
171
    """Server for the TransportTraceDecorator for testing with."""
 
172
 
 
173
    def get_decorator_class(self):
 
174
        from bzrlib.transport import trace
 
175
        return trace.TransportTraceDecorator
 
176
 
 
177
 
 
178
class UnlistableServer(DecoratorServer):
 
179
    """Server for the UnlistableTransportDecorator for testing with."""
 
180
 
 
181
    def get_decorator_class(self):
 
182
        from bzrlib.transport import unlistable
 
183
        return unlistable.UnlistableTransportDecorator
 
184
 
 
185
 
 
186
class TestingPathFilteringServer(pathfilter.PathFilteringServer):
 
187
 
 
188
    def __init__(self):
 
189
        """TestingPathFilteringServer is not usable until start_server
 
190
        is called."""
 
191
 
 
192
    def start_server(self, backing_server=None):
 
193
        """Setup the Chroot on backing_server."""
 
194
        if backing_server is not None:
 
195
            self.backing_transport = transport.get_transport(
 
196
                backing_server.get_url())
 
197
        else:
 
198
            self.backing_transport = transport.get_transport('.')
 
199
        self.backing_transport.clone('added-by-filter').ensure_base()
 
200
        self.filter_func = lambda x: 'added-by-filter/' + x
 
201
        super(TestingPathFilteringServer, self).start_server()
 
202
 
 
203
    def get_bogus_url(self):
 
204
        raise NotImplementedError
 
205
 
 
206
 
 
207
class TestingChrootServer(chroot.ChrootServer):
 
208
 
 
209
    def __init__(self):
 
210
        """TestingChrootServer is not usable until start_server is called."""
 
211
        super(TestingChrootServer, self).__init__(None)
 
212
 
 
213
    def start_server(self, backing_server=None):
 
214
        """Setup the Chroot on backing_server."""
 
215
        if backing_server is not None:
 
216
            self.backing_transport = transport.get_transport(
 
217
                backing_server.get_url())
 
218
        else:
 
219
            self.backing_transport = transport.get_transport('.')
 
220
        super(TestingChrootServer, self).start_server()
 
221
 
 
222
    def get_bogus_url(self):
 
223
        raise NotImplementedError
 
224
 
 
225
 
 
226
class SmartTCPServer_for_testing(server.SmartTCPServer):
 
227
    """Server suitable for use by transport tests.
 
228
 
 
229
    This server is backed by the process's cwd.
 
230
    """
 
231
 
 
232
    def __init__(self, thread_name_suffix=''):
 
233
        super(SmartTCPServer_for_testing, self).__init__(None)
 
234
        self.client_path_extra = None
 
235
        self.thread_name_suffix = thread_name_suffix
 
236
 
 
237
    def get_backing_transport(self, backing_transport_server):
 
238
        """Get a backing transport from a server we are decorating."""
 
239
        return transport.get_transport(backing_transport_server.get_url())
 
240
 
 
241
    def start_server(self, backing_transport_server=None,
 
242
              client_path_extra='/extra/'):
 
243
        """Set up server for testing.
 
244
 
 
245
        :param backing_transport_server: backing server to use.  If not
 
246
            specified, a LocalURLServer at the current working directory will
 
247
            be used.
 
248
        :param client_path_extra: a path segment starting with '/' to append to
 
249
            the root URL for this server.  For instance, a value of '/foo/bar/'
 
250
            will mean the root of the backing transport will be published at a
 
251
            URL like `bzr://127.0.0.1:nnnn/foo/bar/`, rather than
 
252
            `bzr://127.0.0.1:nnnn/`.  Default value is `extra`, so that tests
 
253
            by default will fail unless they do the necessary path translation.
 
254
        """
 
255
        if not client_path_extra.startswith('/'):
 
256
            raise ValueError(client_path_extra)
 
257
        from bzrlib.transport.chroot import ChrootServer
 
258
        if backing_transport_server is None:
 
259
            backing_transport_server = LocalURLServer()
 
260
        self.chroot_server = ChrootServer(
 
261
            self.get_backing_transport(backing_transport_server))
 
262
        self.chroot_server.start_server()
 
263
        self.backing_transport = transport.get_transport(
 
264
            self.chroot_server.get_url())
 
265
        self.root_client_path = self.client_path_extra = client_path_extra
 
266
        self.start_background_thread(self.thread_name_suffix)
 
267
 
 
268
    def stop_server(self):
 
269
        self.stop_background_thread()
 
270
        self.chroot_server.stop_server()
 
271
 
 
272
    def get_url(self):
 
273
        url = super(SmartTCPServer_for_testing, self).get_url()
 
274
        return url[:-1] + self.client_path_extra
 
275
 
 
276
    def get_bogus_url(self):
 
277
        """Return a URL which will fail to connect"""
 
278
        return 'bzr://127.0.0.1:1/'
 
279
 
 
280
 
 
281
class ReadonlySmartTCPServer_for_testing(SmartTCPServer_for_testing):
 
282
    """Get a readonly server for testing."""
 
283
 
 
284
    def get_backing_transport(self, backing_transport_server):
 
285
        """Get a backing transport from a server we are decorating."""
 
286
        url = 'readonly+' + backing_transport_server.get_url()
 
287
        return transport.get_transport(url)
 
288
 
 
289
 
 
290
class SmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing):
 
291
    """A variation of SmartTCPServer_for_testing that limits the client to
 
292
    using RPCs in protocol v2 (i.e. bzr <= 1.5).
 
293
    """
 
294
 
 
295
    def get_url(self):
 
296
        url = super(SmartTCPServer_for_testing_v2_only, self).get_url()
 
297
        url = 'bzr-v2://' + url[len('bzr://'):]
 
298
        return url
 
299
 
 
300
 
 
301
class ReadonlySmartTCPServer_for_testing_v2_only(
 
302
    SmartTCPServer_for_testing_v2_only):
 
303
    """Get a readonly server for testing."""
 
304
 
 
305
    def get_backing_transport(self, backing_transport_server):
 
306
        """Get a backing transport from a server we are decorating."""
 
307
        url = 'readonly+' + backing_transport_server.get_url()
 
308
        return transport.get_transport(url)
 
309
 
 
310
 
 
311
 
 
312