1
# Bazaar -- distributed version control
3
# Copyright (C) 2005 by Canonical Ltd
1
# Copyright (C) 2005, 2006 Canonical Ltd
5
3
# This program is free software; you can redistribute it and/or modify
6
4
# it under the terms of the GNU General Public License as published by
17
15
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
19
17
from cStringIO import StringIO
21
from stat import (S_ISREG, S_ISDIR, S_ISLNK, ST_MODE, ST_SIZE,
22
S_ISCHR, S_ISBLK, S_ISFIFO, S_ISSOCK)
26
from bzrlib.lazy_import import lazy_import
27
lazy_import(globals(), """
29
from datetime import datetime
21
31
from ntpath import (abspath as _nt_abspath,
24
34
realpath as _nt_realpath,
25
35
splitdrive as _nt_splitdrive,
28
from os import listdir
33
from shutil import copyfile
35
from stat import (S_ISREG, S_ISDIR, S_ISLNK, ST_MODE, ST_SIZE,
36
S_ISCHR, S_ISBLK, S_ISFIFO, S_ISSOCK)
45
from tempfile import (
45
from bzrlib.errors import (BzrError,
46
BzrBadParameterNotUnicode,
51
from bzrlib.symbol_versioning import (deprecated_function,
56
from bzrlib.symbol_versioning import (
53
60
from bzrlib.trace import mutter
122
129
return _mapper(_lstat(f).st_mode)
123
130
except OSError, e:
124
131
if getattr(e, 'errno', None) == errno.ENOENT:
125
raise bzrlib.errors.NoSuchFile(f)
132
raise errors.NoSuchFile(f)
159
166
if e.errno == errno.ENOENT:
162
raise BzrError("lstat/stat of (%r): %r" % (f, e))
169
raise errors.BzrError("lstat/stat of (%r): %r" % (f, e))
165
172
def fancy_rename(old, new, rename_func, unlink_func):
186
193
file_existed = False
188
195
rename_func(new, tmp_name)
189
except (NoSuchFile,), e:
196
except (errors.NoSuchFile,), e:
191
198
except IOError, e:
192
199
# RBC 20060103 abstraction leakage: the paramiko SFTP clients rename
221
228
# choke on a Unicode string containing a relative path if
222
229
# os.getcwd() returns a non-sys.getdefaultencoding()-encoded
224
_fs_enc = sys.getfilesystemencoding()
231
_fs_enc = sys.getfilesystemencoding() or 'utf-8'
225
232
def _posix_abspath(path):
226
233
# jam 20060426 rather than encoding to fsencoding
227
234
# copy posixpath.abspath, but use os.getcwdu instead
302
309
pathjoin = os.path.join
303
310
normpath = os.path.normpath
304
311
getcwd = os.getcwdu
305
mkdtemp = tempfile.mkdtemp
306
312
rename = os.rename
307
313
dirname = os.path.dirname
308
314
basename = os.path.basename
309
rmtree = shutil.rmtree
315
split = os.path.split
316
splitext = os.path.splitext
317
# These were already imported into local scope
318
# mkdtemp = tempfile.mkdtemp
319
# rmtree = shutil.rmtree
311
321
MIN_ABS_PATHLENGTH = 1
326
336
"""Error handler for shutil.rmtree function [for win32]
327
337
Helps to remove files and dirs marked as read-only.
329
type_, value = excinfo[:2]
339
exception = excinfo[1]
330
340
if function in (os.remove, os.rmdir) \
331
and type_ == OSError \
332
and value.errno == errno.EACCES:
333
bzrlib.osutils.make_writable(path)
341
and isinstance(exception, OSError) \
342
and exception.errno == errno.EACCES:
366
376
mutter('encoding stdout as sys.stdin encoding %r', output_encoding)
368
378
mutter('encoding stdout as sys.stdout encoding %r', output_encoding)
379
if output_encoding == 'cp0':
380
# invalid encoding (cp0 means 'no codepage' on Windows)
381
output_encoding = bzrlib.user_encoding
382
mutter('cp0 is invalid encoding.'
383
' encoding stdout as bzrlib.user_encoding %r', output_encoding)
386
codecs.lookup(output_encoding)
388
sys.stderr.write('bzr: warning:'
389
' unknown terminal encoding %s.\n'
390
' Using encoding %s instead.\n'
391
% (output_encoding, bzrlib.user_encoding)
393
output_encoding = bzrlib.user_encoding
369
395
return output_encoding
441
467
The empty string as a dir name is taken as top-of-tree and matches
444
>>> is_inside('src', pathjoin('src', 'foo.c'))
446
>>> is_inside('src', 'srccontrol')
448
>>> is_inside('src', pathjoin('src', 'a', 'a', 'a', 'foo.c'))
450
>>> is_inside('foo.c', 'foo.c')
452
>>> is_inside('foo.c', '')
454
>>> is_inside('', 'foo.c')
457
470
# XXX: Most callers of this can actually do something smarter by
458
471
# looking at the inventory
555
568
def local_time_offset(t=None):
556
569
"""Return offset of local zone from GMT, either at present or at time t."""
557
# python2.3 localtime() can't take None
561
if time.localtime(t).tm_isdst and time.daylight:
564
return -time.timezone
572
offset = datetime.fromtimestamp(t) - datetime.utcfromtimestamp(t)
573
return offset.days * 86400 + offset.seconds
567
576
def format_date(t, offset=0, timezone='original', date_fmt=None,
581
590
tt = time.localtime(t)
582
591
offset = local_time_offset(t)
584
raise BzrError("unsupported timezone format %r" % timezone,
585
['options are "utc", "original", "local"'])
593
raise errors.BzrError("unsupported timezone format %r" % timezone,
594
['options are "utc", "original", "local"'])
586
595
if date_fmt is None:
587
596
date_fmt = "%a %Y-%m-%d %H:%M:%S"
659
668
except (NotImplementedError, AttributeError):
660
669
# If python doesn't have os.urandom, or it doesn't work,
661
670
# then try to first pull random data from /dev/urandom
662
if os.path.exists("/dev/urandom"):
663
672
rand_bytes = file('/dev/urandom', 'rb').read
664
673
# Otherwise, use this hack as a last resort
674
except (IOError, OSError):
666
675
# not well seeded, but better than nothing
667
676
def rand_bytes(n):
690
699
## decomposition (might be too tricksy though.)
692
701
def splitpath(p):
693
"""Turn string into list of parts.
699
>>> splitpath('a/./b')
701
>>> splitpath('a/.b')
703
>>> splitpath('a/../b')
704
Traceback (most recent call last):
706
BzrError: sorry, '..' not allowed in path
708
assert isinstance(p, types.StringTypes)
702
"""Turn string into list of parts."""
703
assert isinstance(p, basestring)
710
705
# split on either delimiter because people might use either on
725
720
assert isinstance(p, list)
727
722
if (f == '..') or (f is None) or (f == ''):
728
raise BzrError("sorry, %r not allowed in path" % f)
723
raise errors.BzrError("sorry, %r not allowed in path" % f)
729
724
return pathjoin(*p)
753
748
def link_or_copy(src, dest):
754
749
"""Hardlink a file, or copy it if it can't be hardlinked."""
755
750
if not hardlinks_good():
751
shutil.copyfile(src, dest)
759
754
os.link(src, dest)
760
755
except (OSError, IOError), e:
761
756
if e.errno != errno.EXDEV:
758
shutil.copyfile(src, dest)
765
760
def delete_any(full_path):
766
761
"""Delete a file or directory."""
846
841
return unicode_or_utf8_string.decode('utf8')
847
842
except UnicodeDecodeError:
848
raise BzrBadParameterNotUnicode(unicode_or_utf8_string)
843
raise errors.BzrBadParameterNotUnicode(unicode_or_utf8_string)
851
846
_platform_normalizes_filenames = False
950
945
if sys.platform != "win32":
952
947
if _validWin32PathRE.match(path) is None:
953
raise IllegalPath(path)
948
raise errors.IllegalPath(path)
956
951
def walkdirs(top, prefix=""):
1073
1068
_cached_user_encoding = None
1076
def get_user_encoding():
1071
def get_user_encoding(use_cache=True):
1077
1072
"""Find out what the preferred user encoding is.
1079
1074
This is generally the encoding that is used for command line parameters
1080
1075
and file contents. This may be different from the terminal encoding
1081
1076
or the filesystem encoding.
1078
:param use_cache: Enable cache for detected encoding.
1079
(This parameter is turned on by default,
1080
and required only for selftesting)
1083
1082
:return: A string defining the preferred user encoding
1085
1084
global _cached_user_encoding
1086
if _cached_user_encoding is not None:
1085
if _cached_user_encoding is not None and use_cache:
1087
1086
return _cached_user_encoding
1089
1088
if sys.platform == 'darwin':
1105
1104
' doesn\'t support the locale set by $LANG (%s)\n'
1106
1105
" Continuing with ascii encoding.\n"
1107
1106
% (e, os.environ.get('LANG')))
1109
if _cached_user_encoding is None:
1110
_cached_user_encoding = 'ascii'
1111
return _cached_user_encoding
1107
user_encoding = 'ascii'
1109
# Windows returns 'cp0' to indicate there is no code page. So we'll just
1110
# treat that as ASCII, and not support printing unicode characters to the
1112
if user_encoding in (None, 'cp0'):
1113
user_encoding = 'ascii'
1117
codecs.lookup(user_encoding)
1119
sys.stderr.write('bzr: warning:'
1120
' unknown encoding %s.'
1121
' Continuing with ascii encoding.\n'
1124
user_encoding = 'ascii'
1127
_cached_user_encoding = user_encoding
1129
return user_encoding
1132
def recv_all(socket, bytes):
1133
"""Receive an exact number of bytes.
1135
Regular Socket.recv() may return less than the requested number of bytes,
1136
dependning on what's in the OS buffer. MSG_WAITALL is not available
1137
on all platforms, but this should work everywhere. This will return
1138
less than the requested amount if the remote end closes.
1140
This isn't optimized and is intended mostly for use in testing.
1143
while len(b) < bytes:
1144
new = socket.recv(bytes - len(b))
1150
def dereference_path(path):
1151
"""Determine the real path to a file.
1153
All parent elements are dereferenced. But the file itself is not
1155
:param path: The original path. May be absolute or relative.
1156
:return: the real path *to* the file
1158
parent, base = os.path.split(path)
1159
# The pathjoin for '.' is a workaround for Python bug #1213894.
1160
# (initial path components aren't dereferenced)
1161
return pathjoin(realpath(pathjoin('.', parent)), base)
added
removed
bzrlib/osutils.py
