← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/memory.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2006-08-17 07:52:09 UTC
  • mfrom: (1910.3.4 trivial)
  • Revision ID: pqm@pqm.ubuntu.com-20060817075209-e85a1f9e05ff8b87
(andrew) Trivial fixes to NotImplemented errors.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/transport/memory.py
20
20
so this is primarily useful for testing.
21
21
"""
22
22
 
 
23
from copy import copy
23
24
import os
24
25
import errno
25
26
import re
27
28
from cStringIO import StringIO
28
29
import warnings
29
30
 
30
 
from bzrlib.errors import (
31
 
    FileExists,
32
 
    LockError,
33
 
    InProcessTransport,
34
 
    NoSuchFile,
35
 
    TransportError,
36
 
    )
 
31
from bzrlib.errors import TransportError, NoSuchFile, FileExists, LockError
37
32
from bzrlib.trace import mutter
38
 
from bzrlib.transport import (
39
 
    AppendBasedFileStream,
40
 
    _file_streams,
41
 
    LateReadError,
42
 
    register_transport,
43
 
    Server,
44
 
    Transport,
45
 
    )
 
33
from bzrlib.transport import (Transport, register_transport, Server)
46
34
import bzrlib.urlutils as urlutils
47
35
 
48
36
 
71
59
        if url[-1] != '/':
72
60
            url = url + '/'
73
61
        super(MemoryTransport, self).__init__(url)
74
 
        split = url.find(':') + 3
75
 
        self._scheme = url[:split]
76
 
        self._cwd = url[split:]
 
62
        self._cwd = url[url.find(':') + 3:]
77
63
        # dictionaries from absolute path to file mode
78
64
        self._dirs = {'/':None}
79
65
        self._files = {}
81
67
 
82
68
    def clone(self, offset=None):
83
69
        """See Transport.clone()."""
84
 
        path = self._combine_paths(self._cwd, offset)
85
 
        if len(path) == 0 or path[-1] != '/':
86
 
            path += '/'
87
 
        url = self._scheme + path
 
70
        if offset is None or offset == '':
 
71
            return copy(self)
 
72
        segments = offset.split('/')
 
73
        cwdsegments = self._cwd.split('/')[:-1]
 
74
        while len(segments):
 
75
            segment = segments.pop(0)
 
76
            if segment == '.':
 
77
                continue
 
78
            if segment == '..':
 
79
                if len(cwdsegments) > 1:
 
80
                    cwdsegments.pop()
 
81
                continue
 
82
            cwdsegments.append(segment)
 
83
        url = self.base[:self.base.find(':') + 3] + '/'.join(cwdsegments) + '/'
88
84
        result = MemoryTransport(url)
89
85
        result._dirs = self._dirs
90
86
        result._files = self._files
102
98
        else:
103
99
            return temp_t.base[:-1]
104
100
 
105
 
    def append_file(self, relpath, f, mode=None):
106
 
        """See Transport.append_file()."""
 
101
    def append(self, relpath, f, mode=None):
 
102
        """See Transport.append()."""
107
103
        _abspath = self._abspath(relpath)
108
104
        self._check_parent(_abspath)
109
105
        orig_content, orig_mode = self._files.get(_abspath, ("", None))
121
117
    def has(self, relpath):
122
118
        """See Transport.has()."""
123
119
        _abspath = self._abspath(relpath)
124
 
        return (_abspath in self._files) or (_abspath in self._dirs)
 
120
        return _abspath in self._files or _abspath in self._dirs
125
121
 
126
122
    def delete(self, relpath):
127
123
        """See Transport.delete()."""
130
126
            raise NoSuchFile(relpath)
131
127
        del self._files[_abspath]
132
128
 
133
 
    def external_url(self):
134
 
        """See bzrlib.transport.Transport.external_url."""
135
 
        # MemoryTransport's are only accessible in-process
136
 
        # so we raise here
137
 
        raise InProcessTransport(self)
138
 
 
139
129
    def get(self, relpath):
140
130
        """See Transport.get()."""
141
131
        _abspath = self._abspath(relpath)
142
132
        if not _abspath in self._files:
143
 
            if _abspath in self._dirs:
144
 
                return LateReadError(relpath)
145
 
            else:
146
 
                raise NoSuchFile(relpath)
 
133
            raise NoSuchFile(relpath)
147
134
        return StringIO(self._files[_abspath][0])
148
135
 
149
 
    def put_file(self, relpath, f, mode=None):
150
 
        """See Transport.put_file()."""
 
136
    def put(self, relpath, f, mode=None):
 
137
        """See Transport.put()."""
151
138
        _abspath = self._abspath(relpath)
152
139
        self._check_parent(_abspath)
153
 
        bytes = f.read()
154
 
        if type(bytes) is not str:
155
 
            # Although not strictly correct, we raise UnicodeEncodeError to be
156
 
            # compatible with other transports.
157
 
            raise UnicodeEncodeError(
158
 
                'undefined', bytes, 0, 1,
159
 
                'put_file must be given a file of bytes, not unicode.')
160
 
        self._files[_abspath] = (bytes, mode)
161
 
        return len(bytes)
 
140
        self._files[_abspath] = (f.read(), mode)
162
141
 
163
142
    def mkdir(self, relpath, mode=None):
164
143
        """See Transport.mkdir()."""
168
147
            raise FileExists(relpath)
169
148
        self._dirs[_abspath]=mode
170
149
 
171
 
    def open_write_stream(self, relpath, mode=None):
172
 
        """See Transport.open_write_stream."""
173
 
        self.put_bytes(relpath, "", mode)
174
 
        result = AppendBasedFileStream(self, relpath)
175
 
        _file_streams[self.abspath(relpath)] = result
176
 
        return result
177
 
 
178
150
    def listable(self):
179
151
        """See Transport.listable."""
180
152
        return True
182
154
    def iter_files_recursive(self):
183
155
        for file in self._files:
184
156
            if file.startswith(self._cwd):
185
 
                yield urlutils.escape(file[len(self._cwd):])
 
157
                yield file[len(self._cwd):]
186
158
    
187
159
    def list_dir(self, relpath):
188
160
        """See Transport.list_dir()."""
190
162
        if _abspath != '/' and _abspath not in self._dirs:
191
163
            raise NoSuchFile(relpath)
192
164
        result = []
193
 
 
194
 
        if not _abspath.endswith('/'):
195
 
            _abspath += '/'
196
 
 
197
 
        for path_group in self._files, self._dirs:
198
 
            for path in path_group:
199
 
                if path.startswith(_abspath):
200
 
                    trailing = path[len(_abspath):]
201
 
                    if trailing and '/' not in trailing:
202
 
                        result.append(trailing)
203
 
        return map(urlutils.escape, result)
 
165
        for path in self._files:
 
166
            if (path.startswith(_abspath) and 
 
167
                path[len(_abspath) + 1:].find('/') == -1 and
 
168
                len(path) > len(_abspath)):
 
169
                result.append(path[len(_abspath) + 1:])
 
170
        for path in self._dirs:
 
171
            if (path.startswith(_abspath) and 
 
172
                path[len(_abspath) + 1:].find('/') == -1 and
 
173
                len(path) > len(_abspath) and
 
174
                path[len(_abspath)] == '/'):
 
175
                result.append(path[len(_abspath) + 1:])
 
176
        return result
204
177
 
205
178
    def rename(self, rel_from, rel_to):
206
179
        """Rename a file or directory; fail if the destination exists"""
229
202
        if _abspath in self._files:
230
203
            self._translate_error(IOError(errno.ENOTDIR, relpath), relpath)
231
204
        for path in self._files:
232
 
            if path.startswith(_abspath + '/'):
 
205
            if path.startswith(_abspath):
233
206
                self._translate_error(IOError(errno.ENOTEMPTY, relpath),
234
207
                                      relpath)
235
208
        for path in self._dirs:
236
 
            if path.startswith(_abspath + '/') and path != _abspath:
 
209
            if path.startswith(_abspath) and path != _abspath:
237
210
                self._translate_error(IOError(errno.ENOTEMPTY, relpath), relpath)
238
211
        if not _abspath in self._dirs:
239
212
            raise NoSuchFile(relpath)
263
236
        relpath = urlutils.unescape(relpath)
264
237
        if relpath.find('..') != -1:
265
238
            raise AssertionError('relpath contains ..')
266
 
        if relpath == '':
267
 
            return '/'
268
 
        if relpath[0] == '/':
269
 
            return relpath
270
239
        if relpath == '.':
271
240
            if (self._cwd == '/'):
272
241
                return self._cwd