1
# Copyright (C) 2005, 2006, 2007, 2008, 2010 Canonical Ltd
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.
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.
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
21
from bzrlib.transport import (
25
from bzrlib.smart import server
28
class TestServer(transport.Server):
29
"""A Transport Server dedicated to tests.
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.
36
Note that these are real servers - they must implement all the things
37
that we want bzr transports to take advantage of.
41
"""Return a url for this server.
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.
49
Subsequent calls will return the same resource.
51
raise NotImplementedError
53
def get_bogus_url(self):
54
"""Return a url for this protocol, that will fail to connect.
56
This may raise NotImplementedError to indicate that this server cannot
59
raise NotImplementedError
62
class LocalURLServer(TestServer):
63
"""A pretend server for local transports, using file:// urls.
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.
69
def start_server(self):
73
"""See Transport.Server.get_url."""
74
return urlutils.local_path_to_url('')
77
class DecoratorServer(TestServer):
78
"""Server for the TransportDecorator for testing with.
80
To use this when subclassing TransportDecorator, override override the
81
get_decorator_class method.
84
def start_server(self, server=None):
85
"""See bzrlib.transport.Server.start_server.
87
:server: decorate the urls given by server. If not provided a
88
LocalServer is created.
90
if server is not None:
91
self._made_server = False
94
self._made_server = True
95
self._server = LocalURLServer()
96
self._server.start_server()
98
def stop_server(self):
100
self._server.stop_server()
102
def get_decorator_class(self):
103
"""Return the class of the decorators we should be constructing."""
104
raise NotImplementedError(self.get_decorator_class)
106
def get_url_prefix(self):
107
"""What URL prefix does this decorator produce?"""
108
return self.get_decorator_class()._get_url_prefix()
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()
115
"""See bzrlib.transport.Server.get_url."""
116
return self.get_url_prefix() + self._server.get_url()
119
class BrokenRenameServer(DecoratorServer):
120
"""Server for the BrokenRenameTransportDecorator for testing with."""
122
def get_decorator_class(self):
123
from bzrlib.transport import brokenrename
124
return brokenrename.BrokenRenameTransportDecorator
127
class FakeNFSServer(DecoratorServer):
128
"""Server for the FakeNFSTransportDecorator for testing with."""
130
def get_decorator_class(self):
131
from bzrlib.transport import fakenfs
132
return fakenfs.FakeNFSTransportDecorator
135
class FakeVFATServer(DecoratorServer):
136
"""A server that suggests connections through FakeVFATTransportDecorator
141
def get_decorator_class(self):
142
from bzrlib.transport import fakevfat
143
return fakevfat.FakeVFATTransportDecorator
146
class LogDecoratorServer(DecoratorServer):
147
"""Server for testing."""
149
def get_decorator_class(self):
150
from bzrlib.transport import log
151
return log.TransportLogDecorator
154
class NoSmartTransportServer(DecoratorServer):
155
"""Server for the NoSmartTransportDecorator for testing with."""
157
def get_decorator_class(self):
158
from bzrlib.transport import nosmart
159
return nosmart.NoSmartTransportDecorator
162
class ReadonlyServer(DecoratorServer):
163
"""Server for the ReadonlyTransportDecorator for testing with."""
165
def get_decorator_class(self):
166
from bzrlib.transport import readonly
167
return readonly.ReadonlyTransportDecorator
170
class TraceServer(DecoratorServer):
171
"""Server for the TransportTraceDecorator for testing with."""
173
def get_decorator_class(self):
174
from bzrlib.transport import trace
175
return trace.TransportTraceDecorator
178
class UnlistableServer(DecoratorServer):
179
"""Server for the UnlistableTransportDecorator for testing with."""
181
def get_decorator_class(self):
182
from bzrlib.transport import unlistable
183
return unlistable.UnlistableTransportDecorator
186
class TestingPathFilteringServer(pathfilter.PathFilteringServer):
189
"""TestingPathFilteringServer is not usable until start_server
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())
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()
203
def get_bogus_url(self):
204
raise NotImplementedError
207
class TestingChrootServer(chroot.ChrootServer):
210
"""TestingChrootServer is not usable until start_server is called."""
211
super(TestingChrootServer, self).__init__(None)
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())
219
self.backing_transport = transport.get_transport('.')
220
super(TestingChrootServer, self).start_server()
222
def get_bogus_url(self):
223
raise NotImplementedError
226
class SmartTCPServer_for_testing(server.SmartTCPServer):
227
"""Server suitable for use by transport tests.
229
This server is backed by the process's cwd.
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
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())
241
def start_server(self, backing_transport_server=None,
242
client_path_extra='/extra/'):
243
"""Set up server for testing.
245
:param backing_transport_server: backing server to use. If not
246
specified, a LocalURLServer at the current working directory will
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.
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)
268
def stop_server(self):
269
self.stop_background_thread()
270
self.chroot_server.stop_server()
273
url = super(SmartTCPServer_for_testing, self).get_url()
274
return url[:-1] + self.client_path_extra
276
def get_bogus_url(self):
277
"""Return a URL which will fail to connect"""
278
return 'bzr://127.0.0.1:1/'
281
class ReadonlySmartTCPServer_for_testing(SmartTCPServer_for_testing):
282
"""Get a readonly server for testing."""
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)
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).
296
url = super(SmartTCPServer_for_testing_v2_only, self).get_url()
297
url = 'bzr-v2://' + url[len('bzr://'):]
301
class ReadonlySmartTCPServer_for_testing_v2_only(
302
SmartTCPServer_for_testing_v2_only):
303
"""Get a readonly server for testing."""
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)
added
removed
bzrlib/tests/test_server.py
