1
# Copyright (C) 2005, 2006 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., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
17
"""Implementation of Transport that uses memory for its storage.
19
The contents of the transport will be lost when the object is discarded,
20
so this is primarily useful for testing.
27
from stat import S_IFREG, S_IFDIR
28
from cStringIO import StringIO
31
from bzrlib.errors import TransportError, NoSuchFile, FileExists, LockError
32
from bzrlib.trace import mutter
33
from bzrlib.transport import (Transport, register_transport, Server)
34
import bzrlib.urlutils as urlutils
38
class MemoryStat(object):
40
def __init__(self, size, is_dir, perms):
45
self.st_mode = S_IFREG | perms
49
self.st_mode = S_IFDIR | perms
52
class MemoryTransport(Transport):
53
"""This is an in memory file system for transient data storage."""
55
def __init__(self, url=""):
56
"""Set the 'base' path where files will be stored."""
61
super(MemoryTransport, self).__init__(url)
62
self._cwd = url[url.find(':') + 3:]
63
# dictionaries from absolute path to file mode
64
self._dirs = {'/':None}
68
def clone(self, offset=None):
69
"""See Transport.clone()."""
70
if offset is None or offset == '':
72
segments = offset.split('/')
73
if offset.startswith('/'):
76
cwdsegments = self._cwd.split('/')[:-1]
78
segment = segments.pop(0)
82
if len(cwdsegments) > 1:
86
cwdsegments.append(segment)
87
url = self.base[:self.base.find(':') + 3] + '/'.join(cwdsegments) + '/'
88
result = MemoryTransport(url)
89
result._dirs = self._dirs
90
result._files = self._files
91
result._locks = self._locks
94
def abspath(self, relpath):
95
"""See Transport.abspath()."""
96
# while a little slow, this is sufficiently fast to not matter in our
97
# current environment - XXX RBC 20060404 move the clone '..' handling
98
# into here and call abspath from clone
99
temp_t = self.clone(relpath)
100
if temp_t.base.count('/') == 3:
103
return temp_t.base[:-1]
105
def append_file(self, relpath, f, mode=None):
106
"""See Transport.append_file()."""
107
_abspath = self._abspath(relpath)
108
self._check_parent(_abspath)
109
orig_content, orig_mode = self._files.get(_abspath, ("", None))
112
self._files[_abspath] = (orig_content + f.read(), mode)
113
return len(orig_content)
115
def _check_parent(self, _abspath):
116
dir = os.path.dirname(_abspath)
118
if not dir in self._dirs:
119
raise NoSuchFile(_abspath)
121
def has(self, relpath):
122
"""See Transport.has()."""
123
_abspath = self._abspath(relpath)
124
return (_abspath in self._files) or (_abspath in self._dirs)
126
def delete(self, relpath):
127
"""See Transport.delete()."""
128
_abspath = self._abspath(relpath)
129
if not _abspath in self._files:
130
raise NoSuchFile(relpath)
131
del self._files[_abspath]
133
def get(self, relpath):
134
"""See Transport.get()."""
135
_abspath = self._abspath(relpath)
136
if not _abspath in self._files:
137
raise NoSuchFile(relpath)
138
return StringIO(self._files[_abspath][0])
140
def put_file(self, relpath, f, mode=None):
141
"""See Transport.put_file()."""
142
_abspath = self._abspath(relpath)
143
self._check_parent(_abspath)
144
self._files[_abspath] = (f.read(), mode)
146
def mkdir(self, relpath, mode=None):
147
"""See Transport.mkdir()."""
148
_abspath = self._abspath(relpath)
149
self._check_parent(_abspath)
150
if _abspath in self._dirs:
151
raise FileExists(relpath)
152
self._dirs[_abspath]=mode
155
"""See Transport.listable."""
158
def iter_files_recursive(self):
159
for file in self._files:
160
if file.startswith(self._cwd):
161
yield urlutils.escape(file[len(self._cwd):])
163
def list_dir(self, relpath):
164
"""See Transport.list_dir()."""
165
_abspath = self._abspath(relpath)
166
if _abspath != '/' and _abspath not in self._dirs:
167
raise NoSuchFile(relpath)
169
for path in self._files:
170
if (path.startswith(_abspath) and
171
path[len(_abspath) + 1:].find('/') == -1 and
172
len(path) > len(_abspath)):
173
result.append(path[len(_abspath) + 1:])
174
for path in self._dirs:
175
if (path.startswith(_abspath) and
176
path[len(_abspath) + 1:].find('/') == -1 and
177
len(path) > len(_abspath) and
178
path[len(_abspath)] == '/'):
179
result.append(path[len(_abspath) + 1:])
180
return map(urlutils.escape, result)
182
def rename(self, rel_from, rel_to):
183
"""Rename a file or directory; fail if the destination exists"""
184
abs_from = self._abspath(rel_from)
185
abs_to = self._abspath(rel_to)
189
elif x.startswith(abs_from + '/'):
190
x = abs_to + x[len(abs_from):]
192
def do_renames(container):
193
for path in container:
194
new_path = replace(path)
196
if new_path in container:
197
raise FileExists(new_path)
198
container[new_path] = container[path]
200
do_renames(self._files)
201
do_renames(self._dirs)
203
def rmdir(self, relpath):
204
"""See Transport.rmdir."""
205
_abspath = self._abspath(relpath)
206
if _abspath in self._files:
207
self._translate_error(IOError(errno.ENOTDIR, relpath), relpath)
208
for path in self._files:
209
if path.startswith(_abspath):
210
self._translate_error(IOError(errno.ENOTEMPTY, relpath),
212
for path in self._dirs:
213
if path.startswith(_abspath) and path != _abspath:
214
self._translate_error(IOError(errno.ENOTEMPTY, relpath), relpath)
215
if not _abspath in self._dirs:
216
raise NoSuchFile(relpath)
217
del self._dirs[_abspath]
219
def stat(self, relpath):
220
"""See Transport.stat()."""
221
_abspath = self._abspath(relpath)
222
if _abspath in self._files:
223
return MemoryStat(len(self._files[_abspath][0]), False,
224
self._files[_abspath][1])
225
elif _abspath in self._dirs:
226
return MemoryStat(0, True, self._dirs[_abspath])
228
raise NoSuchFile(_abspath)
230
def lock_read(self, relpath):
231
"""See Transport.lock_read()."""
232
return _MemoryLock(self._abspath(relpath), self)
234
def lock_write(self, relpath):
235
"""See Transport.lock_write()."""
236
return _MemoryLock(self._abspath(relpath), self)
238
def _abspath(self, relpath):
239
"""Generate an internal absolute path."""
240
relpath = urlutils.unescape(relpath)
241
if relpath.find('..') != -1:
242
raise AssertionError('relpath contains ..')
243
if relpath[0] == '/':
246
if (self._cwd == '/'):
248
return self._cwd[:-1]
249
if relpath.endswith('/'):
250
relpath = relpath[:-1]
251
if relpath.startswith('./'):
252
relpath = relpath[2:]
253
return self._cwd + relpath
256
class _MemoryLock(object):
257
"""This makes a lock."""
259
def __init__(self, path, transport):
260
assert isinstance(transport, MemoryTransport)
262
self.transport = transport
263
if self.path in self.transport._locks:
264
raise LockError('File %r already locked' % (self.path,))
265
self.transport._locks[self.path] = self
268
# Should this warn, or actually try to cleanup?
270
warnings.warn("MemoryLock %r not explicitly unlocked" % (self.path,))
274
del self.transport._locks[self.path]
275
self.transport = None
278
class MemoryServer(Server):
279
"""Server for the MemoryTransport for testing with."""
282
"""See bzrlib.transport.Server.setUp."""
283
self._dirs = {'/':None}
286
self._scheme = "memory+%s:///" % id(self)
287
def memory_factory(url):
288
result = MemoryTransport(url)
289
result._dirs = self._dirs
290
result._files = self._files
291
result._locks = self._locks
293
register_transport(self._scheme, memory_factory)
296
"""See bzrlib.transport.Server.tearDown."""
297
# unregister this server
300
"""See bzrlib.transport.Server.get_url."""
304
def get_test_permutations():
305
"""Return the permutations to be used in testing."""
306
return [(MemoryTransport, MemoryServer),
added
removed
bzrlib/transport/memory.py
