"""
import sys, exceptions, os, stat, tempfile, time, binascii
+from collections import namedtuple
+from errno import ENOENT
from twisted.python import log
def shutdown(self):
remove(self.name)
-class NamedTemporaryDirectory:
- """
- This calls tempfile.mkdtemp(), stores the name of the dir in
- self.name, and rmrf's the dir when it gets garbage collected or
- "shutdown()".
- """
- def __init__(self, cleanup=True, *args, **kwargs):
- """ If cleanup, then the directory will be rmrf'ed when the object is shutdown. """
- self.cleanup = cleanup
- self.name = tempfile.mkdtemp(*args, **kwargs)
-
- def __repr__(self):
- return "<%s instance at %x %s>" % (self.__class__.__name__, id(self), self.name)
-
- def __str__(self):
- return self.__repr__()
-
- def __del__(self):
- try:
- self.shutdown()
- except:
- import traceback
- traceback.print_exc()
-
- def shutdown(self):
- if self.cleanup and hasattr(self, 'name'):
- rm_dir(self.name)
-
class EncryptedTemporaryFile:
# not implemented: next, readline, readlines, xreadlines, writelines
except ImportError:
pass
-def abspath_expanduser_unicode(path, base=None):
+def abspath_expanduser_unicode(path, base=None, long_path=True):
"""
Return the absolute version of a path. If 'base' is given and 'path' is relative,
the path will be expanded relative to 'base'.
'path' must be a Unicode string. 'base', if given, must be a Unicode string
corresponding to an absolute path as returned by a previous call to
abspath_expanduser_unicode.
+ On Windows, the result will be a long path unless long_path is given as False.
"""
if not isinstance(path, unicode):
raise AssertionError("paths must be Unicode strings")
- if base is not None:
+ if base is not None and long_path:
precondition_abspath(base)
path = expanduser(path)
if _getfullpathname:
- # On Windows, os.path.isabs will return True for paths without a drive letter,
+ # On Windows, os.path.isabs will incorrectly return True
+ # for paths without a drive letter (that are not UNC paths),
# e.g. "\\". See <http://bugs.python.org/issue1669539>.
try:
- path = _getfullpathname(path or u".")
+ if base is None:
+ path = _getfullpathname(path or u".")
+ else:
+ path = _getfullpathname(os.path.join(base, path))
except OSError:
pass
# there is always at least one Unicode path component.
path = os.path.normpath(path)
- if sys.platform == "win32":
+ if sys.platform == "win32" and long_path:
path = to_windows_long_path(path)
return path
have_GetDiskFreeSpaceExW = False
if sys.platform == "win32":
- from ctypes import WINFUNCTYPE, windll, POINTER, byref, c_ulonglong, create_unicode_buffer
+ from ctypes import WINFUNCTYPE, windll, POINTER, byref, c_ulonglong, create_unicode_buffer, \
+ get_last_error
from ctypes.wintypes import BOOL, DWORD, LPCWSTR, LPWSTR
- # <http://msdn.microsoft.com/en-us/library/ms679360%28v=VS.85%29.aspx>
- GetLastError = WINFUNCTYPE(DWORD)(("GetLastError", windll.kernel32))
-
# <http://msdn.microsoft.com/en-us/library/windows/desktop/ms683188%28v=vs.85%29.aspx>
- GetEnvironmentVariableW = WINFUNCTYPE(DWORD, LPCWSTR, LPWSTR, DWORD)(
- ("GetEnvironmentVariableW", windll.kernel32))
+ GetEnvironmentVariableW = WINFUNCTYPE(
+ DWORD,
+ LPCWSTR, LPWSTR, DWORD,
+ use_last_error=True
+ )(("GetEnvironmentVariableW", windll.kernel32))
try:
# <http://msdn.microsoft.com/en-us/library/aa383742%28v=VS.85%29.aspx>
PULARGE_INTEGER = POINTER(c_ulonglong)
# <http://msdn.microsoft.com/en-us/library/aa364937%28VS.85%29.aspx>
- GetDiskFreeSpaceExW = WINFUNCTYPE(BOOL, LPCWSTR, PULARGE_INTEGER, PULARGE_INTEGER, PULARGE_INTEGER)(
- ("GetDiskFreeSpaceExW", windll.kernel32))
+ GetDiskFreeSpaceExW = WINFUNCTYPE(
+ BOOL,
+ LPCWSTR, PULARGE_INTEGER, PULARGE_INTEGER, PULARGE_INTEGER,
+ use_last_error=True
+ )(("GetDiskFreeSpaceExW", windll.kernel32))
have_GetDiskFreeSpaceExW = True
except Exception:
def windows_expanduser(path):
if not path.startswith('~'):
return path
- home_drive = windows_getenv(u'HOMEDRIVE')
- home_path = windows_getenv(u'HOMEPATH')
+
+ home_dir = windows_getenv(u'USERPROFILE')
+ if home_dir is None:
+ home_drive = windows_getenv(u'HOMEDRIVE')
+ home_path = windows_getenv(u'HOMEPATH')
+ if home_drive is None or home_path is None:
+ raise OSError("Could not find home directory: neither %USERPROFILE% nor (%HOMEDRIVE% and %HOMEPATH%) are set.")
+ home_dir = os.path.join(home_drive, home_path)
+
if path == '~':
- return os.path.join(home_drive, home_path)
+ return home_dir
elif path.startswith('~/') or path.startswith('~\\'):
- return os.path.join(home_drive, home_path, path[2 :])
+ return os.path.join(home_dir, path[2 :])
else:
return path
+# <https://msdn.microsoft.com/en-us/library/windows/desktop/ms681382%28v=vs.85%29.aspx>
+ERROR_ENVVAR_NOT_FOUND = 203
+
def windows_getenv(name):
# Based on <http://stackoverflow.com/questions/2608200/problems-with-umlauts-in-python-appdata-environvent-variable/2608368#2608368>,
- # with improved error handling.
+ # with improved error handling. Returns None if there is no enivronment variable of the given name.
if not isinstance(name, unicode):
raise AssertionError("name must be Unicode")
n = GetEnvironmentVariableW(name, None, 0)
# GetEnvironmentVariableW returns DWORD, so n cannot be negative.
if n == 0:
- err = GetLastError()
+ err = get_last_error()
+ if err == ERROR_ENVVAR_NOT_FOUND:
+ return None
raise OSError("Windows error %d attempting to read size of environment variable %r"
% (err, name))
if n == 1:
buf = create_unicode_buffer(u'\0'*n)
retval = GetEnvironmentVariableW(name, buf, n)
if retval == 0:
- err = GetLastError()
+ err = get_last_error()
+ if err == ERROR_ENVVAR_NOT_FOUND:
+ return None
raise OSError("Windows error %d attempting to read environment variable %r"
% (err, name))
if retval >= n:
byref(n_free_for_root))
if retval == 0:
raise OSError("Windows error %d attempting to get disk statistics for %r"
- % (GetLastError(), whichdir))
+ % (get_last_error(), whichdir))
free_for_nonroot = n_free_for_nonroot.value
total = n_total.value
free_for_root = n_free_for_root.value
except EnvironmentError:
log.msg("OS call to get disk statistics failed")
return 0
+
+
+if sys.platform == "win32":
+ from ctypes.wintypes import BOOL, HANDLE, DWORD, LPCWSTR, LPVOID, WinError, get_last_error
+
+ # <http://msdn.microsoft.com/en-us/library/aa363858%28v=vs.85%29.aspx>
+ CreateFileW = WINFUNCTYPE(HANDLE, LPCWSTR, DWORD, DWORD, LPVOID, DWORD, DWORD, HANDLE) \
+ (("CreateFileW", windll.kernel32))
+
+ GENERIC_WRITE = 0x40000000
+ FILE_SHARE_READ = 0x00000001
+ FILE_SHARE_WRITE = 0x00000002
+ OPEN_EXISTING = 3
+ INVALID_HANDLE_VALUE = 0xFFFFFFFF
+
+ # <http://msdn.microsoft.com/en-us/library/aa364439%28v=vs.85%29.aspx>
+ FlushFileBuffers = WINFUNCTYPE(BOOL, HANDLE)(("FlushFileBuffers", windll.kernel32))
+
+ # <http://msdn.microsoft.com/en-us/library/ms724211%28v=vs.85%29.aspx>
+ CloseHandle = WINFUNCTYPE(BOOL, HANDLE)(("CloseHandle", windll.kernel32))
+
+ # <http://social.msdn.microsoft.com/forums/en-US/netfxbcl/thread/4465cafb-f4ed-434f-89d8-c85ced6ffaa8/>
+ def flush_volume(path):
+ drive = os.path.splitdrive(os.path.realpath(path))[0]
+
+ hVolume = CreateFileW(u"\\\\.\\" + drive,
+ GENERIC_WRITE,
+ FILE_SHARE_READ | FILE_SHARE_WRITE,
+ None,
+ OPEN_EXISTING,
+ 0,
+ None
+ )
+ if hVolume == INVALID_HANDLE_VALUE:
+ raise WinError()
+
+ if FlushFileBuffers(hVolume) == 0:
+ raise WinError()
+
+ CloseHandle(hVolume)
+else:
+ def flush_volume(path):
+ # use sync()?
+ pass
+
+
+class ConflictError(Exception):
+ pass
+
+class UnableToUnlinkReplacementError(Exception):
+ pass
+
+def reraise(wrapper):
+ _, exc, tb = sys.exc_info()
+ wrapper_exc = wrapper("%s: %s" % (exc.__class__.__name__, exc))
+ raise wrapper_exc.__class__, wrapper_exc, tb
+
+if sys.platform == "win32":
+ from ctypes import WINFUNCTYPE, windll, WinError, get_last_error
+ from ctypes.wintypes import BOOL, DWORD, LPCWSTR, LPVOID
+
+ # <https://msdn.microsoft.com/en-us/library/windows/desktop/aa365512%28v=vs.85%29.aspx>
+ ReplaceFileW = WINFUNCTYPE(
+ BOOL,
+ LPCWSTR, LPCWSTR, LPCWSTR, DWORD, LPVOID, LPVOID,
+ use_last_error=True
+ )(("ReplaceFileW", windll.kernel32))
+
+ REPLACEFILE_IGNORE_MERGE_ERRORS = 0x00000002
+
+ def rename_no_overwrite(source_path, dest_path):
+ os.rename(source_path, dest_path)
+
+ def replace_file(replaced_path, replacement_path, backup_path):
+ precondition_abspath(replaced_path)
+ precondition_abspath(replacement_path)
+ precondition_abspath(backup_path)
+
+ r = ReplaceFileW(replaced_path, replacement_path, backup_path,
+ REPLACEFILE_IGNORE_MERGE_ERRORS, None, None)
+ if r == 0:
+ # The UnableToUnlinkReplacementError case does not happen on Windows;
+ # all errors should be treated as signalling a conflict.
+ err = get_last_error()
+ raise ConflictError("WinError: %s" % (WinError(err)))
+else:
+ def rename_no_overwrite(source_path, dest_path):
+ # link will fail with EEXIST if there is already something at dest_path.
+ os.link(source_path, dest_path)
+ try:
+ os.unlink(source_path)
+ except EnvironmentError:
+ reraise(UnableToUnlinkReplacementError)
+
+ def replace_file(replaced_path, replacement_path, backup_path):
+ precondition_abspath(replaced_path)
+ precondition_abspath(replacement_path)
+ precondition_abspath(backup_path)
+
+ if not os.path.exists(replacement_path):
+ raise ConflictError("Replacement file not found: %r" % (replacement_path,))
+
+ try:
+ os.rename(replaced_path, backup_path)
+ except OSError as e:
+ if e.errno != ENOENT:
+ raise
+ try:
+ rename_no_overwrite(replacement_path, replaced_path)
+ except EnvironmentError:
+ reraise(ConflictError)
+
+PathInfo = namedtuple('PathInfo', 'isdir isfile islink exists size mtime ctime')
+
+def get_pathinfo(path_u, now=None):
+ try:
+ statinfo = os.lstat(path_u)
+ mode = statinfo.st_mode
+ return PathInfo(isdir =stat.S_ISDIR(mode),
+ isfile=stat.S_ISREG(mode),
+ islink=stat.S_ISLNK(mode),
+ exists=True,
+ size =statinfo.st_size,
+ mtime =statinfo.st_mtime,
+ ctime =statinfo.st_ctime,
+ )
+ except OSError as e:
+ if e.errno == ENOENT:
+ if now is None:
+ now = time.time()
+ return PathInfo(isdir =False,
+ isfile=False,
+ islink=False,
+ exists=False,
+ size =None,
+ mtime =now,
+ ctime =now,
+ )
+ raise
projects / tahoe-lafs / tahoe-lafs.git / blobdiff