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 (
24
from bzrlib.smart import server
27
class TestServer(transport.Server):
28
"""A Transport Server dedicated to tests.
30
The TestServer interface provides a server for a given transport. We use
31
these servers as loopback testing tools. For any given transport the
32
Servers it provides must either allow writing, or serve the contents
33
of os.getcwdu() at the time start_server is called.
35
Note that these are real servers - they must implement all the things
36
that we want bzr transports to take advantage of.
40
"""Return a url for this server.
42
If the transport does not represent a disk directory (i.e. it is
43
a database like svn, or a memory only transport, it should return
44
a connection to a newly established resource for this Server.
45
Otherwise it should return a url that will provide access to the path
46
that was os.getcwdu() when start_server() was called.
48
Subsequent calls will return the same resource.
50
raise NotImplementedError
52
def get_bogus_url(self):
53
"""Return a url for this protocol, that will fail to connect.
55
This may raise NotImplementedError to indicate that this server cannot
58
raise NotImplementedError
61
class LocalURLServer(TestServer):
62
"""A pretend server for local transports, using file:// urls.
64
Of course no actual server is required to access the local filesystem, so
65
this just exists to tell the test code how to get to it.
68
def start_server(self):
72
"""See Transport.Server.get_url."""
73
return urlutils.local_path_to_url('')
76
class MemoryServer(TestServer):
77
"""Server for the MemoryTransport for testing with."""
79
def start_server(self):
80
self._dirs = {'/':None}
83
self._scheme = "memory+%s:///" % id(self)
84
def memory_factory(url):
85
from bzrlib.transport import memory
86
result = memory.MemoryTransport(url)
87
result._dirs = self._dirs
88
result._files = self._files
89
result._locks = self._locks
91
self._memory_factory = memory_factory
92
transport.register_transport(self._scheme, self._memory_factory)
94
def stop_server(self):
95
# unregister this server
96
transport.unregister_transport(self._scheme, self._memory_factory)
99
"""See bzrlib.transport.Server.get_url."""
103
class DecoratorServer(TestServer):
104
"""Server for the TransportDecorator for testing with.
106
To use this when subclassing TransportDecorator, override override the
107
get_decorator_class method.
110
def start_server(self, server=None):
111
"""See bzrlib.transport.Server.start_server.
113
:server: decorate the urls given by server. If not provided a
114
LocalServer is created.
116
if server is not None:
117
self._made_server = False
118
self._server = server
120
self._made_server = True
121
self._server = LocalURLServer()
122
self._server.start_server()
124
def stop_server(self):
125
if self._made_server:
126
self._server.stop_server()
128
def get_decorator_class(self):
129
"""Return the class of the decorators we should be constructing."""
130
raise NotImplementedError(self.get_decorator_class)
132
def get_url_prefix(self):
133
"""What URL prefix does this decorator produce?"""
134
return self.get_decorator_class()._get_url_prefix()
136
def get_bogus_url(self):
137
"""See bzrlib.transport.Server.get_bogus_url."""
138
return self.get_url_prefix() + self._server.get_bogus_url()
141
"""See bzrlib.transport.Server.get_url."""
142
return self.get_url_prefix() + self._server.get_url()
145
class BrokenRenameServer(DecoratorServer):
146
"""Server for the BrokenRenameTransportDecorator for testing with."""
148
def get_decorator_class(self):
149
from bzrlib.transport import brokenrename
150
return brokenrename.BrokenRenameTransportDecorator
153
class FakeNFSServer(DecoratorServer):
154
"""Server for the FakeNFSTransportDecorator for testing with."""
156
def get_decorator_class(self):
157
from bzrlib.transport import fakenfs
158
return fakenfs.FakeNFSTransportDecorator
161
class FakeVFATServer(DecoratorServer):
162
"""A server that suggests connections through FakeVFATTransportDecorator
167
def get_decorator_class(self):
168
from bzrlib.transport import fakevfat
169
return fakevfat.FakeVFATTransportDecorator
172
class LogDecoratorServer(DecoratorServer):
173
"""Server for testing."""
175
def get_decorator_class(self):
176
from bzrlib.transport import log
177
return log.TransportLogDecorator
180
class NoSmartTransportServer(DecoratorServer):
181
"""Server for the NoSmartTransportDecorator for testing with."""
183
def get_decorator_class(self):
184
from bzrlib.transport import nosmart
185
return nosmart.NoSmartTransportDecorator
188
class ReadonlyServer(DecoratorServer):
189
"""Server for the ReadonlyTransportDecorator for testing with."""
191
def get_decorator_class(self):
192
from bzrlib.transport import readonly
193
return readonly.ReadonlyTransportDecorator
196
class TraceServer(DecoratorServer):
197
"""Server for the TransportTraceDecorator for testing with."""
199
def get_decorator_class(self):
200
from bzrlib.transport import trace
201
return trace.TransportTraceDecorator
204
class UnlistableServer(DecoratorServer):
205
"""Server for the UnlistableTransportDecorator for testing with."""
207
def get_decorator_class(self):
208
from bzrlib.transport import unlistable
209
return unlistable.UnlistableTransportDecorator
212
class TestingPathFilteringServer(pathfilter.PathFilteringServer):
215
"""TestingChrootServer is not usable until start_server is called."""
217
def start_server(self, backing_server=None):
218
"""Setup the Chroot on backing_server."""
219
if backing_server is not None:
220
self.backing_transport = transport.get_transport(
221
backing_server.get_url())
223
self.backing_transport = transport.get_transport('.')
224
self.backing_transport.clone('added-by-filter').ensure_base()
225
self.filter_func = lambda x: 'added-by-filter/' + x
226
super(TestingPathFilteringServer, self).start_server()
229
class SmartTCPServer_for_testing(server.SmartTCPServer):
230
"""Server suitable for use by transport tests.
232
This server is backed by the process's cwd.
235
def __init__(self, thread_name_suffix=''):
236
super(SmartTCPServer_for_testing, self).__init__(None)
237
self.client_path_extra = None
238
self.thread_name_suffix = thread_name_suffix
240
def get_backing_transport(self, backing_transport_server):
241
"""Get a backing transport from a server we are decorating."""
242
return transport.get_transport(backing_transport_server.get_url())
244
def start_server(self, backing_transport_server=None,
245
client_path_extra='/extra/'):
246
"""Set up server for testing.
248
:param backing_transport_server: backing server to use. If not
249
specified, a LocalURLServer at the current working directory will
251
:param client_path_extra: a path segment starting with '/' to append to
252
the root URL for this server. For instance, a value of '/foo/bar/'
253
will mean the root of the backing transport will be published at a
254
URL like `bzr://127.0.0.1:nnnn/foo/bar/`, rather than
255
`bzr://127.0.0.1:nnnn/`. Default value is `extra`, so that tests
256
by default will fail unless they do the necessary path translation.
258
if not client_path_extra.startswith('/'):
259
raise ValueError(client_path_extra)
260
from bzrlib.transport.chroot import ChrootServer
261
if backing_transport_server is None:
262
backing_transport_server = LocalURLServer()
263
self.chroot_server = ChrootServer(
264
self.get_backing_transport(backing_transport_server))
265
self.chroot_server.start_server()
266
self.backing_transport = transport.get_transport(
267
self.chroot_server.get_url())
268
self.root_client_path = self.client_path_extra = client_path_extra
269
self.start_background_thread(self.thread_name_suffix)
271
def stop_server(self):
272
self.stop_background_thread()
273
self.chroot_server.stop_server()
276
url = super(SmartTCPServer_for_testing, self).get_url()
277
return url[:-1] + self.client_path_extra
279
def get_bogus_url(self):
280
"""Return a URL which will fail to connect"""
281
return 'bzr://127.0.0.1:1/'
284
class ReadonlySmartTCPServer_for_testing(SmartTCPServer_for_testing):
285
"""Get a readonly server for testing."""
287
def get_backing_transport(self, backing_transport_server):
288
"""Get a backing transport from a server we are decorating."""
289
url = 'readonly+' + backing_transport_server.get_url()
290
return transport.get_transport(url)
293
class SmartTCPServer_for_testing_v2_only(SmartTCPServer_for_testing):
294
"""A variation of SmartTCPServer_for_testing that limits the client to
295
using RPCs in protocol v2 (i.e. bzr <= 1.5).
299
url = super(SmartTCPServer_for_testing_v2_only, self).get_url()
300
url = 'bzr-v2://' + url[len('bzr://'):]
304
class ReadonlySmartTCPServer_for_testing_v2_only(
305
SmartTCPServer_for_testing_v2_only):
306
"""Get a readonly server for testing."""
308
def get_backing_transport(self, backing_transport_server):
309
"""Get a backing transport from a server we are decorating."""
310
url = 'readonly+' + backing_transport_server.get_url()
311
return transport.get_transport(url)
added
removed
bzrlib/tests/test_server.py
