← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/transport/local.py

  • Committer: Robert Collins
  • Date: 2007-07-04 08:08:13 UTC
  • mfrom: (2572 +trunk)
  • mto: This revision was merged to the branch mainline in revision 2587.
  • Revision ID: robertc@robertcollins.net-20070704080813-wzebx0r88fvwj5rq
Merge bzr.dev.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/transport/local.py
19
19
This is a fairly thin wrapper on regular file IO.
20
20
"""
21
21
 
 
22
import os
 
23
from stat import ST_MODE, S_ISDIR, ST_SIZE, S_IMODE
 
24
import sys
 
25
 
 
26
from bzrlib.lazy_import import lazy_import
 
27
lazy_import(globals(), """
22
28
import errno
23
 
import os
24
29
import shutil
25
 
import sys
26
 
from stat import ST_MODE, S_ISDIR, ST_SIZE, S_IMODE
27
 
import tempfile
28
30
 
29
31
from bzrlib import (
30
32
    atomicfile,
31
33
    osutils,
32
34
    urlutils,
 
35
    symbol_versioning,
33
36
    )
34
 
from bzrlib.osutils import (abspath, realpath, normpath, pathjoin, rename,
35
 
                            check_legal_path, rmtree)
36
 
from bzrlib.symbol_versioning import warn
37
37
from bzrlib.trace import mutter
38
 
from bzrlib.transport import LateReadError, Transport, Server
 
38
from bzrlib.transport import LateReadError
 
39
""")
 
40
 
 
41
from bzrlib.transport import Transport, Server
39
42
 
40
43
 
41
44
_append_flags = os.O_CREAT | os.O_APPEND | os.O_WRONLY | osutils.O_BINARY
48
51
    def __init__(self, base):
49
52
        """Set the base path where files will be stored."""
50
53
        if not base.startswith('file://'):
51
 
            warn("Instantiating LocalTransport with a filesystem path"
 
54
            symbol_versioning.warn(
 
55
                "Instantiating LocalTransport with a filesystem path"
52
56
                " is deprecated as of bzr 0.8."
53
57
                " Please use bzrlib.transport.get_transport()"
54
58
                " or pass in a file:// url.",
72
76
        if offset is None:
73
77
            return LocalTransport(self.base)
74
78
        else:
75
 
            return LocalTransport(self.abspath(offset))
 
79
            abspath = self.abspath(offset)
 
80
            if abspath == 'file://':
 
81
                # fix upwalk for UNC path
 
82
                # when clone from //HOST/path updir recursively
 
83
                # we should stop at least at //HOST part
 
84
                abspath = self.base
 
85
            return LocalTransport(abspath)
76
86
 
77
87
    def _abspath(self, relative_reference):
78
88
        """Return a path for use in os calls.
91
101
        assert isinstance(relpath, basestring), (type(relpath), relpath)
92
102
        # jam 20060426 Using normpath on the real path, because that ensures
93
103
        #       proper handling of stuff like
94
 
        path = normpath(pathjoin(self._local_base, urlutils.unescape(relpath)))
 
104
        path = osutils.normpath(osutils.pathjoin(
 
105
                    self._local_base, urlutils.unescape(relpath)))
95
106
        return urlutils.local_path_to_url(path)
96
107
 
97
108
    def local_abspath(self, relpath):
147
158
        path = relpath
148
159
        try:
149
160
            path = self._abspath(relpath)
150
 
            check_legal_path(path)
 
161
            osutils.check_legal_path(path)
151
162
            fp = atomicfile.AtomicFile(path, 'wb', new_mode=mode)
152
163
        except (IOError, OSError),e:
153
164
            self._translate_error(e, path)
167
178
        path = relpath
168
179
        try:
169
180
            path = self._abspath(relpath)
170
 
            check_legal_path(path)
 
181
            osutils.check_legal_path(path)
171
182
            fp = atomicfile.AtomicFile(path, 'wb', new_mode=mode)
172
183
        except (IOError, OSError),e:
173
184
            self._translate_error(e, path)
368
379
 
369
380
        try:
370
381
            # this version will delete the destination if necessary
371
 
            rename(path_from, path_to)
 
382
            osutils.rename(path_from, path_to)
372
383
        except (IOError, OSError),e:
373
384
            # TODO: What about path_to?
374
385
            self._translate_error(e, path_from)
472
483
            return True
473
484
 
474
485
 
475
 
class LocalRelpathServer(Server):
476
 
    """A pretend server for local transports, using relpaths."""
477
 
 
478
 
    def get_url(self):
479
 
        """See Transport.Server.get_url."""
480
 
        return "."
481
 
 
482
 
 
483
 
class LocalAbspathServer(Server):
484
 
    """A pretend server for local transports, using absolute paths."""
485
 
 
486
 
    def get_url(self):
487
 
        """See Transport.Server.get_url."""
488
 
        return os.path.abspath("")
 
486
class EmulatedWin32LocalTransport(LocalTransport):
 
487
    """Special transport for testing Win32 [UNC] paths on non-windows"""
 
488
 
 
489
    def __init__(self, base):
 
490
        if base[-1] != '/':
 
491
            base = base + '/'
 
492
        super(LocalTransport, self).__init__(base)
 
493
        self._local_base = urlutils._win32_local_path_from_url(base)
 
494
 
 
495
    def abspath(self, relpath):
 
496
        assert isinstance(relpath, basestring), (type(relpath), relpath)
 
497
        path = osutils.normpath(osutils.pathjoin(
 
498
                    self._local_base, urlutils.unescape(relpath)))
 
499
        return urlutils._win32_local_path_to_url(path)
 
500
 
 
501
    def clone(self, offset=None):
 
502
        """Return a new LocalTransport with root at self.base + offset
 
503
        Because the local filesystem does not require a connection, 
 
504
        we can just return a new object.
 
505
        """
 
506
        if offset is None:
 
507
            return EmulatedWin32LocalTransport(self.base)
 
508
        else:
 
509
            abspath = self.abspath(offset)
 
510
            if abspath == 'file://':
 
511
                # fix upwalk for UNC path
 
512
                # when clone from //HOST/path updir recursively
 
513
                # we should stop at least at //HOST part
 
514
                abspath = self.base
 
515
            return EmulatedWin32LocalTransport(abspath)
489
516
 
490
517
 
491
518
class LocalURLServer(Server):
492
 
    """A pretend server for local transports, using file:// urls."""
 
519
    """A pretend server for local transports, using file:// urls.
 
520
    
 
521
    Of course no actual server is required to access the local filesystem, so
 
522
    this just exists to tell the test code how to get to it.
 
523
    """
 
524
 
 
525
    def setUp(self):
 
526
        """Setup the server to service requests.
 
527
        
 
528
        :param decorated_transport: ignored by this implementation.
 
529
        """
493
530
 
494
531
    def get_url(self):
495
532
        """See Transport.Server.get_url."""
498
535
 
499
536
def get_test_permutations():
500
537
    """Return the permutations to be used in testing."""
501
 
    return [(LocalTransport, LocalRelpathServer),
502
 
            (LocalTransport, LocalAbspathServer),
 
538
    return [
503
539
            (LocalTransport, LocalURLServer),
504
540
            ]