19
19
This is a fairly thin wrapper on regular file IO.
23
from stat import ST_MODE, S_ISDIR, ST_SIZE, S_IMODE
26
from bzrlib.lazy_import import lazy_import
27
lazy_import(globals(), """
26
from stat import ST_MODE, S_ISDIR, ST_SIZE, S_IMODE
29
31
from bzrlib import (
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
41
from bzrlib.transport import Transport, Server
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.",
73
77
return LocalTransport(self.base)
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
85
return LocalTransport(abspath)
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)
97
108
def local_abspath(self, relpath):
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)
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)
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)
475
class LocalRelpathServer(Server):
476
"""A pretend server for local transports, using relpaths."""
479
"""See Transport.Server.get_url."""
483
class LocalAbspathServer(Server):
484
"""A pretend server for local transports, using absolute paths."""
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"""
489
def __init__(self, base):
492
super(LocalTransport, self).__init__(base)
493
self._local_base = urlutils._win32_local_path_from_url(base)
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)
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.
507
return EmulatedWin32LocalTransport(self.base)
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
515
return EmulatedWin32LocalTransport(abspath)
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.
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.
526
"""Setup the server to service requests.
528
:param decorated_transport: ignored by this implementation.
494
531
def get_url(self):
495
532
"""See Transport.Server.get_url."""
499
536
def get_test_permutations():
500
537
"""Return the permutations to be used in testing."""
501
return [(LocalTransport, LocalRelpathServer),
502
(LocalTransport, LocalAbspathServer),
503
539
(LocalTransport, LocalURLServer),