13
13
# You should have received a copy of the GNU General Public License
14
14
# along with this program; if not, write to the Free Software
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
15
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
17
17
"""Implementation of Transport that uses memory for its storage.
20
20
so this is primarily useful for testing.
23
from __future__ import absolute_import
26
27
from stat import S_IFREG, S_IFDIR
27
28
from cStringIO import StringIO
30
from bzrlib.errors import TransportError, NoSuchFile, FileExists, LockError
31
from bzrlib.trace import mutter
32
from bzrlib.transport import (Transport, register_transport, Server)
33
import bzrlib.urlutils as urlutils
34
from bzrlib.errors import (
40
from bzrlib.transport import (
41
AppendBasedFileStream,
69
80
def clone(self, offset=None):
70
81
"""See Transport.clone()."""
71
path = self._combine_paths(self._cwd, offset)
82
path = urlutils.URL._combine_paths(self._cwd, offset)
72
83
if len(path) == 0 or path[-1] != '/':
74
85
url = self._scheme + path
75
result = MemoryTransport(url)
86
result = self.__class__(url)
76
87
result._dirs = self._dirs
77
88
result._files = self._files
78
89
result._locks = self._locks
117
128
raise NoSuchFile(relpath)
118
129
del self._files[_abspath]
131
def external_url(self):
132
"""See bzrlib.transport.Transport.external_url."""
133
# MemoryTransport's are only accessible in-process
135
raise InProcessTransport(self)
120
137
def get(self, relpath):
121
138
"""See Transport.get()."""
122
139
_abspath = self._abspath(relpath)
123
140
if not _abspath in self._files:
124
raise NoSuchFile(relpath)
141
if _abspath in self._dirs:
142
return LateReadError(relpath)
144
raise NoSuchFile(relpath)
125
145
return StringIO(self._files[_abspath][0])
127
147
def put_file(self, relpath, f, mode=None):
136
156
'undefined', bytes, 0, 1,
137
157
'put_file must be given a file of bytes, not unicode.')
138
158
self._files[_abspath] = (bytes, mode)
140
161
def mkdir(self, relpath, mode=None):
141
162
"""See Transport.mkdir()."""
145
166
raise FileExists(relpath)
146
167
self._dirs[_abspath]=mode
169
def open_write_stream(self, relpath, mode=None):
170
"""See Transport.open_write_stream."""
171
self.put_bytes(relpath, "", mode)
172
result = AppendBasedFileStream(self, relpath)
173
_file_streams[self.abspath(relpath)] = result
148
176
def listable(self):
149
177
"""See Transport.listable."""
213
241
"""See Transport.stat()."""
214
242
_abspath = self._abspath(relpath)
215
243
if _abspath in self._files:
216
return MemoryStat(len(self._files[_abspath][0]), False,
244
return MemoryStat(len(self._files[_abspath][0]), False,
217
245
self._files[_abspath][1])
218
246
elif _abspath in self._dirs:
219
247
return MemoryStat(0, True, self._dirs[_abspath])
231
259
def _abspath(self, relpath):
232
260
"""Generate an internal absolute path."""
233
261
relpath = urlutils.unescape(relpath)
234
if relpath.find('..') != -1:
235
raise AssertionError('relpath contains ..')
238
if relpath[0] == '/':
262
if relpath[:1] == '/':
241
if (self._cwd == '/'):
243
return self._cwd[:-1]
244
if relpath.endswith('/'):
245
relpath = relpath[:-1]
246
if relpath.startswith('./'):
247
relpath = relpath[2:]
248
return self._cwd + relpath
264
cwd_parts = self._cwd.split('/')
265
rel_parts = relpath.split('/')
267
for i in cwd_parts + rel_parts:
270
raise ValueError("illegal relpath %r under %r"
271
% (relpath, self._cwd))
273
elif i == '.' or i == '':
277
return '/' + '/'.join(r)
251
280
class _MemoryLock(object):
252
281
"""This makes a lock."""
254
283
def __init__(self, path, transport):
255
assert isinstance(transport, MemoryTransport)
257
285
self.transport = transport
258
286
if self.path in self.transport._locks:
259
287
raise LockError('File %r already locked' % (self.path,))
260
288
self.transport._locks[self.path] = self
263
# Should this warn, or actually try to cleanup?
265
warnings.warn("MemoryLock %r not explicitly unlocked" % (self.path,))
268
290
def unlock(self):
269
291
del self.transport._locks[self.path]
270
292
self.transport = None
273
class MemoryServer(Server):
295
class MemoryServer(transport.Server):
274
296
"""Server for the MemoryTransport for testing with."""
277
"""See bzrlib.transport.Server.setUp."""
298
def start_server(self):
278
299
self._dirs = {'/':None}
281
302
self._scheme = "memory+%s:///" % id(self)
282
303
def memory_factory(url):
283
result = MemoryTransport(url)
304
from bzrlib.transport import memory
305
result = memory.MemoryTransport(url)
284
306
result._dirs = self._dirs
285
307
result._files = self._files
286
308
result._locks = self._locks
288
register_transport(self._scheme, memory_factory)
310
self._memory_factory = memory_factory
311
transport.register_transport(self._scheme, self._memory_factory)
291
"""See bzrlib.transport.Server.tearDown."""
313
def stop_server(self):
292
314
# unregister this server
315
transport.unregister_transport(self._scheme, self._memory_factory)
294
317
def get_url(self):
295
318
"""See bzrlib.transport.Server.get_url."""
296
319
return self._scheme
321
def get_bogus_url(self):
322
raise NotImplementedError
299
325
def get_test_permutations():
300
326
"""Return the permutations to be used in testing."""