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:
pass
-def open_or_create(fname, binarymode=True):
- try:
- return open(fname, binarymode and "r+b" or "r+")
- except EnvironmentError:
- return open(fname, binarymode and "w+b" or "w+")
-
def du(basedir):
size = 0
remove_if_possible(dest)
os.rename(source, dest)
-def write(path, data):
- wf = open(path, "wb")
+def write_atomically(target, contents, mode="b"):
+ f = open(target+".tmp", "w"+mode)
+ try:
+ f.write(contents)
+ finally:
+ f.close()
+ move_into_place(target+".tmp", target)
+
+def write(path, data, mode="wb"):
+ wf = open(path, mode)
try:
wf.write(data)
finally:
finally:
rf.close()
-def put_file(pathname, inf):
+def put_file(path, inf):
+ precondition_abspath(path)
+
# TODO: create temporary file and move into place?
- outf = open(os.path.expanduser(pathname), "wb")
+ outf = open(path, "wb")
try:
while True:
data = inf.read(32768)
outf.close()
+def precondition_abspath(path):
+ if not isinstance(path, unicode):
+ raise AssertionError("an abspath must be a Unicode string")
+
+ if sys.platform == "win32":
+ # This intentionally doesn't view absolute paths starting with a drive specification, or
+ # paths relative to the current drive, as acceptable.
+ if not path.startswith("\\\\"):
+ raise AssertionError("an abspath should be normalized using abspath_expanduser_unicode")
+ else:
+ # This intentionally doesn't view the path '~' or paths starting with '~/' as acceptable.
+ if not os.path.isabs(path):
+ raise AssertionError("an abspath should be normalized using abspath_expanduser_unicode")
+
# Work around <http://bugs.python.org/issue3426>. This code is adapted from
# <http://svn.python.org/view/python/trunk/Lib/ntpath.py?revision=78247&view=markup>
# with some simplifications.
except ImportError:
pass
-def abspath_expanduser_unicode(path):
- """Return the absolute version of a path."""
- assert isinstance(path, unicode), path
+def abspath_expanduser_unicode(path, base=None):
+ """
+ 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.
+ """
+ if not isinstance(path, unicode):
+ raise AssertionError("paths must be Unicode strings")
+ if base is not None:
+ precondition_abspath(base)
- path = os.path.expanduser(path)
+ 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
if not os.path.isabs(path):
- path = os.path.join(os.getcwdu(), path)
+ if base is None:
+ path = os.path.join(os.getcwdu(), path)
+ else:
+ path = os.path.join(base, path)
# We won't hit <http://bugs.python.org/issue5827> because
# there is always at least one Unicode path component.
- return os.path.normpath(path)
+ path = os.path.normpath(path)
-windows = False
-try:
- import win32api, win32con
-except ImportError:
- pass
-else:
- windows = True
- # <http://msdn.microsoft.com/en-us/library/ms680621%28VS.85%29.aspx>
- win32api.SetErrorMode(win32con.SEM_FAILCRITICALERRORS |
- win32con.SEM_NOOPENFILEERRORBOX)
+ if sys.platform == "win32":
+ path = to_windows_long_path(path)
+
+ return path
+
+def to_windows_long_path(path):
+ # '/' is normally a perfectly valid path component separator in Windows.
+ # However, when using the "\\?\" syntax it is not recognized, so we
+ # replace it with '\' here.
+ path = path.replace(u"/", u"\\")
+
+ # Note that other normalizations such as removing '.' and '..' should
+ # be done outside this function.
+
+ if path.startswith(u"\\\\?\\") or path.startswith(u"\\\\.\\"):
+ return path
+ elif path.startswith(u"\\\\"):
+ return u"\\\\?\\UNC\\" + path[2 :]
+ else:
+ return u"\\\\?\\" + path
+
+
+have_GetDiskFreeSpaceExW = False
+if sys.platform == "win32":
+ 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/windows/desktop/ms683188%28v=vs.85%29.aspx>
+ 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,
+ use_last_error=True
+ )(("GetDiskFreeSpaceExW", windll.kernel32))
+
+ have_GetDiskFreeSpaceExW = True
+ except Exception:
+ import traceback
+ traceback.print_exc()
+
+def expanduser(path):
+ # os.path.expanduser is hopelessly broken for Unicode paths on Windows (ticket #1674).
+ if sys.platform == "win32":
+ return windows_expanduser(path)
+ else:
+ return os.path.expanduser(path)
+
+def windows_expanduser(path):
+ if not path.startswith('~'):
+ return path
+
+ 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 home_dir
+ elif path.startswith('~/') or path.startswith('~\\'):
+ 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. 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 = 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:
+ # Avoid an ambiguity between a zero-length string and an error in the return value of the
+ # call to GetEnvironmentVariableW below.
+ return u""
+
+ buf = create_unicode_buffer(u'\0'*n)
+ retval = GetEnvironmentVariableW(name, buf, n)
+ if retval == 0:
+ 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:
+ raise OSError("Unexpected result %d (expected less than %d) from GetEnvironmentVariableW attempting to read environment variable %r"
+ % (retval, n, name))
+
+ return buf.value
def get_disk_stats(whichdir, reserved_space=0):
"""Return disk statistics for the storage disk, in the form of a dict
filesystem as reserved_space.
"""
- if windows:
- # For Windows systems, where os.statvfs is not available, use GetDiskFreeSpaceEx.
- # <http://docs.activestate.com/activepython/2.5/pywin32/win32api__GetDiskFreeSpaceEx_meth.html>
- #
- # Although the docs say that the argument should be the root directory
- # of a disk, GetDiskFreeSpaceEx actually accepts any path on that disk
- # (like its Win32 equivalent).
-
- (free_for_nonroot, total, free_for_root) = win32api.GetDiskFreeSpaceEx(whichdir)
+ if have_GetDiskFreeSpaceExW:
+ # If this is a Windows system and GetDiskFreeSpaceExW is available, use it.
+ # (This might put up an error dialog unless
+ # SetErrorMode(SEM_FAILCRITICALERRORS | SEM_NOOPENFILEERRORBOX) has been called,
+ # which we do in allmydata.windows.fixups.initialize().)
+
+ n_free_for_nonroot = c_ulonglong(0)
+ n_total = c_ulonglong(0)
+ n_free_for_root = c_ulonglong(0)
+ retval = GetDiskFreeSpaceExW(whichdir, byref(n_free_for_nonroot),
+ byref(n_total),
+ byref(n_free_for_root))
+ if retval == 0:
+ raise OSError("Windows error %d attempting to get disk statistics for %r"
+ % (get_last_error(), whichdir))
+ free_for_nonroot = n_free_for_nonroot.value
+ total = n_total.value
+ free_for_root = n_free_for_root.value
else:
# For Unix-like systems.
# <http://docs.python.org/library/os.html#os.statvfs>
used = total - free_for_root
avail = max(free_for_nonroot - reserved_space, 0)
- return { 'total': total, 'free_for_root': free_for_root,
+ return { 'total': total,
+ 'free_for_root': free_for_root,
'free_for_nonroot': free_for_nonroot,
- 'used': used, 'avail': avail, }
+ 'used': used,
+ 'avail': avail,
+ }
def get_available_space(whichdir, reserved_space):
"""Returns available space for share storage in bytes, or None if no
projects / tahoe-lafs / tahoe-lafs.git / blobdiff