← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/memory.py

  • Committer: Tarmac
  • Author(s): Vincent Ladeuil
  • Date: 2017-01-30 14:42:05 UTC
  • mfrom: (6620.1.1 trunk)
  • Revision ID: tarmac-20170130144205-r8fh2xpmiuxyozpv
Merge  2.7 into trunk including fix for bug #1657238 [r=vila]

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/transport/memory.py
1
 
# Copyright (C) 2005, 2006 Canonical Ltd
 
1
# Copyright (C) 2005-2011, 2016 Canonical Ltd
2
2
#
3
3
# This program is free software; you can redistribute it and/or modify
4
4
# it under the terms of the GNU General Public License as published by
20
20
so this is primarily useful for testing.
21
21
"""
22
22
 
 
23
from __future__ import absolute_import
 
24
 
23
25
import os
24
26
import errno
25
 
import re
26
27
from stat import S_IFREG, S_IFDIR
27
28
from cStringIO import StringIO
28
 
import warnings
29
29
 
 
30
from bzrlib import (
 
31
    transport,
 
32
    urlutils,
 
33
    )
30
34
from bzrlib.errors import (
31
35
    FileExists,
32
36
    LockError,
33
37
    InProcessTransport,
34
38
    NoSuchFile,
35
 
    TransportError,
36
39
    )
37
 
from bzrlib.trace import mutter
38
40
from bzrlib.transport import (
39
41
    AppendBasedFileStream,
40
42
    _file_streams,
41
43
    LateReadError,
42
 
    register_transport,
43
 
    Server,
44
 
    Transport,
45
44
    )
46
 
import bzrlib.urlutils as urlutils
47
45
 
48
46
 
49
47
 
61
59
            self.st_mode = S_IFDIR | perms
62
60
 
63
61
 
64
 
class MemoryTransport(Transport):
 
62
class MemoryTransport(transport.Transport):
65
63
    """This is an in memory file system for transient data storage."""
66
64
 
67
65
    def __init__(self, url=""):
81
79
 
82
80
    def clone(self, offset=None):
83
81
        """See Transport.clone()."""
84
 
        path = self._combine_paths(self._cwd, offset)
 
82
        path = urlutils.URL._combine_paths(self._cwd, offset)
85
83
        if len(path) == 0 or path[-1] != '/':
86
84
            path += '/'
87
85
        url = self._scheme + path
150
148
        """See Transport.put_file()."""
151
149
        _abspath = self._abspath(relpath)
152
150
        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)
 
151
        raw_bytes = f.read()
 
152
        self._files[_abspath] = (raw_bytes, mode)
 
153
        return len(raw_bytes)
162
154
 
163
155
    def mkdir(self, relpath, mode=None):
164
156
        """See Transport.mkdir()."""
289
281
            raise LockError('File %r already locked' % (self.path,))
290
282
        self.transport._locks[self.path] = self
291
283
 
292
 
    def __del__(self):
293
 
        # Should this warn, or actually try to cleanup?
294
 
        if self.transport:
295
 
            warnings.warn("MemoryLock %r not explicitly unlocked" % (self.path,))
296
 
            self.unlock()
297
 
 
298
284
    def unlock(self):
299
285
        del self.transport._locks[self.path]
300
286
        self.transport = None
301
287
 
302
288
 
303
 
class MemoryServer(Server):
 
289
class MemoryServer(transport.Server):
304
290
    """Server for the MemoryTransport for testing with."""
305
291
 
306
 
    def setUp(self):
307
 
        """See bzrlib.transport.Server.setUp."""
 
292
    def start_server(self):
308
293
        self._dirs = {'/':None}
309
294
        self._files = {}
310
295
        self._locks = {}
311
296
        self._scheme = "memory+%s:///" % id(self)
312
297
        def memory_factory(url):
313
 
            result = MemoryTransport(url)
 
298
            from bzrlib.transport import memory
 
299
            result = memory.MemoryTransport(url)
314
300
            result._dirs = self._dirs
315
301
            result._files = self._files
316
302
            result._locks = self._locks
317
303
            return result
318
 
        register_transport(self._scheme, memory_factory)
 
304
        self._memory_factory = memory_factory
 
305
        transport.register_transport(self._scheme, self._memory_factory)
319
306
 
320
 
    def tearDown(self):
321
 
        """See bzrlib.transport.Server.tearDown."""
 
307
    def stop_server(self):
322
308
        # unregister this server
 
309
        transport.unregister_transport(self._scheme, self._memory_factory)
323
310
 
324
311
    def get_url(self):
325
312
        """See bzrlib.transport.Server.get_url."""
326
313
        return self._scheme
327
314
 
 
315
    def get_bogus_url(self):
 
316
        raise NotImplementedError
 
317
 
328
318
 
329
319
def get_test_permutations():
330
320
    """Return the permutations to be used in testing."""