Index: Lib/test/test_tempfile.py =================================================================== --- Lib/test/test_tempfile.py (revision 0) +++ Lib/test/test_tempfile.py (revision 0) @@ -0,0 +1,682 @@ +# tempfile.py unit tests. + +import tempfile +import os +import sys +import re +import errno +import warnings + +import unittest +from test import test_support + +warnings.filterwarnings("ignore", + category=RuntimeWarning, + message="mktemp", module=__name__) + +if hasattr(os, 'stat'): + import stat + has_stat = 1 +else: + has_stat = 0 + +has_textmode = (tempfile._text_openflags != tempfile._bin_openflags) +has_spawnl = hasattr(os, 'spawnl') + +# TEST_FILES may need to be tweaked for systems depending on the maximum +# number of files that can be opened at one time (see ulimit -n) +if sys.platform == 'mac': + TEST_FILES = 32 +elif sys.platform in ('openbsd3', 'openbsd4'): + TEST_FILES = 48 +else: + TEST_FILES = 100 + +# This is organized as one test for each chunk of code in tempfile.py, +# in order of their appearance in the file. Testing which requires +# threads is not done here. + +# Common functionality. +class TC(unittest.TestCase): + + str_check = re.compile(r"[a-zA-Z0-9_-]{2,8}$") + + def failOnException(self, what, ei=None): + if ei is None: + ei = sys.exc_info() + self.fail("%s raised %s: %s" % (what, ei[0], ei[1])) + + def nameCheck(self, name, dir, pre, suf): + (ndir, nbase) = os.path.split(name) + npre = nbase[:len(pre)] + nsuf = nbase[len(nbase)-len(suf):] + + # check for equality of the absolute paths! + self.assertEqual(os.path.abspath(ndir), os.path.abspath(dir), + "file '%s' not in directory '%s'" % (name, dir)) + self.assertEqual(npre, pre, + "file '%s' does not begin with '%s'" % (nbase, pre)) + self.assertEqual(nsuf, suf, + "file '%s' does not end with '%s'" % (nbase, suf)) + + nbase = nbase[len(pre):len(nbase)-len(suf)] + self.assert_(self.str_check.match(nbase), + "random string '%s' does not match /^[a-zA-Z0-9_-]{2,8}$/" + % nbase) + +test_classes = [] + +class test_exports(TC): + def test_exports(self): + # There are no surprising symbols in the tempfile module + dict = tempfile.__dict__ + + expected = { + "NamedTemporaryFile" : 1, + "TemporaryFile" : 1, + "mkstemp" : 1, + "mkdtemp" : 1, + "mktemp" : 1, + "TMP_MAX" : 1, + "gettempprefix" : 1, + "gettempdir" : 1, + "tempdir" : 1, + "template" : 1 + } + + unexp = [] + for key in dict: + if key[0] != '_' and key not in expected: + unexp.append(key) + self.failUnless(len(unexp) == 0, + "unexpected keys: %s" % unexp) + +test_classes.append(test_exports) + + +class test__RandomNameSequence(TC): + """Test the internal iterator object _RandomNameSequence.""" + + def setUp(self): + self.r = tempfile._RandomNameSequence() + + def test_get_six_char_str(self): + # _RandomNameSequence returns a six-character string + s = self.r.next() + self.nameCheck(s, '', '', '') + + def test_many(self): + # _RandomNameSequence returns no duplicate strings (stochastic) + + dict = {} + r = self.r + for i in xrange(TEST_FILES): + s = r.next() + self.nameCheck(s, '', '', '') + self.failIf(s in dict) + dict[s] = 1 + + def test_supports_iter(self): + # _RandomNameSequence supports the iterator protocol + + i = 0 + r = self.r + try: + for s in r: + i += 1 + if i == 20: + break + except: + failOnException("iteration") + +test_classes.append(test__RandomNameSequence) + + +class test__candidate_tempdir_list(TC): + """Test the internal function _candidate_tempdir_list.""" + + def test_nonempty_list(self): + # _candidate_tempdir_list returns a nonempty list of strings + + cand = tempfile._candidate_tempdir_list() + + self.failIf(len(cand) == 0) + for c in cand: + self.assert_(isinstance(c, basestring), + "%s is not a string" % c) + + def test_wanted_dirs(self): + # _candidate_tempdir_list contains the expected directories + + # Make sure the interesting environment variables are all set. + added = [] + try: + for envname in 'TMPDIR', 'TEMP', 'TMP': + dirname = os.getenv(envname) + if not dirname: + os.environ[envname] = os.path.abspath(envname) + added.append(envname) + + cand = tempfile._candidate_tempdir_list() + + for envname in 'TMPDIR', 'TEMP', 'TMP': + dirname = os.getenv(envname) + if not dirname: raise ValueError + self.assert_(dirname in cand) + + try: + dirname = os.getcwd() + except (AttributeError, os.error): + dirname = os.curdir + + self.assert_(dirname in cand) + + # Not practical to try to verify the presence of OS-specific + # paths in this list. + finally: + for p in added: + del os.environ[p] + +test_classes.append(test__candidate_tempdir_list) + + +# We test _get_default_tempdir by testing gettempdir. + + +class test__get_candidate_names(TC): + """Test the internal function _get_candidate_names.""" + + def test_retval(self): + # _get_candidate_names returns a _RandomNameSequence object + obj = tempfile._get_candidate_names() + self.assert_(isinstance(obj, tempfile._RandomNameSequence)) + + def test_same_thing(self): + # _get_candidate_names always returns the same object + a = tempfile._get_candidate_names() + b = tempfile._get_candidate_names() + + self.assert_(a is b) + +test_classes.append(test__get_candidate_names) + + +class test__mkstemp_inner(TC): + """Test the internal function _mkstemp_inner.""" + + class mkstemped: + _bflags = tempfile._bin_openflags + _tflags = tempfile._text_openflags + + def __init__(self, dir, pre, suf, bin): + if bin: flags = self._bflags + else: flags = self._tflags + + # XXX: Assign _close/_unlink here, instead of as a class + # var. submit back to CPython + # XXX: close doesn't work here + #self._close = os.close + self._unlink = os.unlink + (self.fd, self.name) = tempfile._mkstemp_inner(dir, pre, suf, flags) + + def write(self, str): + os.write(self.fd, str) + + def __del__(self): + self._close(self.fd) + self._unlink(self.name) + + def do_create(self, dir=None, pre="", suf="", bin=1): + if dir is None: + dir = tempfile.gettempdir() + try: + file = self.mkstemped(dir, pre, suf, bin) + except: + self.failOnException("_mkstemp_inner") + + self.nameCheck(file.name, dir, pre, suf) + return file + + def test_basic(self): + # _mkstemp_inner can create files + self.do_create().write("blat") + self.do_create(pre="a").write("blat") + self.do_create(suf="b").write("blat") + self.do_create(pre="a", suf="b").write("blat") + self.do_create(pre="aa", suf=".txt").write("blat") + + def test_basic_many(self): + # _mkstemp_inner can create many files (stochastic) + extant = range(TEST_FILES) + for i in extant: + extant[i] = self.do_create(pre="aa") + + def test_choose_directory(self): + # _mkstemp_inner can create files in a user-selected directory + dir = tempfile.mkdtemp() + try: + self.do_create(dir=dir).write("blat") + finally: + os.rmdir(dir) + + def test_file_mode(self): + # _mkstemp_inner creates files with the proper mode + if not has_stat: + return # ugh, can't use TestSkipped. + + file = self.do_create() + mode = stat.S_IMODE(os.stat(file.name).st_mode) + expected = 0600 + if sys.platform in ('win32', 'os2emx', 'mac'): + # There's no distinction among 'user', 'group' and 'world'; + # replicate the 'user' bits. + user = expected >> 6 + expected = user * (1 + 8 + 64) + self.assertEqual(mode, expected) + + def test_noinherit(self): + # _mkstemp_inner file handles are not inherited by child processes + if not has_spawnl: + return # ugh, can't use TestSkipped. + + if test_support.verbose: + v="v" + else: + v="q" + + file = self.do_create() + fd = "%d" % file.fd + + try: + me = __file__ + except NameError: + me = sys.argv[0] + + # We have to exec something, so that FD_CLOEXEC will take + # effect. The core of this test is therefore in + # tf_inherit_check.py, which see. + tester = os.path.join(os.path.dirname(os.path.abspath(me)), + "tf_inherit_check.py") + + # On Windows a spawn* /path/ with embedded spaces shouldn't be quoted, + # but an arg with embedded spaces should be decorated with double + # quotes on each end + if sys.platform in ('win32'): + decorated = '"%s"' % sys.executable + tester = '"%s"' % tester + else: + decorated = sys.executable + + retval = os.spawnl(os.P_WAIT, sys.executable, decorated, tester, v, fd) + self.failIf(retval < 0, + "child process caught fatal signal %d" % -retval) + self.failIf(retval > 0, "child process reports failure %d"%retval) + + def test_textmode(self): + # _mkstemp_inner can create files in text mode + if not has_textmode: + return # ugh, can't use TestSkipped. + + self.do_create(bin=0).write("blat\n") + # XXX should test that the file really is a text file + +# XXX: Java lacks support for file descriptors +#test_classes.append(test__mkstemp_inner) + + +class test_gettempprefix(TC): + """Test gettempprefix().""" + + def test_sane_template(self): + # gettempprefix returns a nonempty prefix string + p = tempfile.gettempprefix() + + self.assert_(isinstance(p, basestring)) + self.assert_(len(p) > 0) + + def test_usable_template(self): + # gettempprefix returns a usable prefix string + + # Create a temp directory, avoiding use of the prefix. + # Then attempt to create a file whose name is + # prefix + 'xxxxxx.xxx' in that directory. + p = tempfile.gettempprefix() + "xxxxxx.xxx" + d = tempfile.mkdtemp(prefix="") + try: + p = os.path.join(d, p) + try: + fp = open(p, 'w+b') + except: + self.failOnException("os.open") + fp.close() + os.unlink(p) + finally: + os.rmdir(d) + +test_classes.append(test_gettempprefix) + + +class test_gettempdir(TC): + """Test gettempdir().""" + + def test_directory_exists(self): + # gettempdir returns a directory which exists + + dir = tempfile.gettempdir() + self.assert_(os.path.isabs(dir) or dir == os.curdir, + "%s is not an absolute path" % dir) + self.assert_(os.path.isdir(dir), + "%s is not a directory" % dir) + + def test_directory_writable(self): + # gettempdir returns a directory writable by the user + + # sneaky: just instantiate a NamedTemporaryFile, which + # defaults to writing into the directory returned by + # gettempdir. + try: + file = tempfile.NamedTemporaryFile() + file.write("blat") + file.close() + except: + self.failOnException("create file in %s" % tempfile.gettempdir()) + + def test_same_thing(self): + # gettempdir always returns the same object + a = tempfile.gettempdir() + b = tempfile.gettempdir() + + self.assert_(a is b) + +test_classes.append(test_gettempdir) + + +class test_mkstemp(TC): + """Test mkstemp().""" + + def do_create(self, dir=None, pre="", suf=""): + if dir is None: + dir = tempfile.gettempdir() + try: + (fd, name) = tempfile.mkstemp(dir=dir, prefix=pre, suffix=suf) + (ndir, nbase) = os.path.split(name) + adir = os.path.abspath(dir) + self.assertEqual(adir, ndir, + "Directory '%s' incorrectly returned as '%s'" % (adir, ndir)) + except: + self.failOnException("mkstemp") + + try: + self.nameCheck(name, dir, pre, suf) + finally: + os.unlink(name) + + def test_basic(self): + # mkstemp can create files + self.do_create() + self.do_create(pre="a") + self.do_create(suf="b") + self.do_create(pre="a", suf="b") + self.do_create(pre="aa", suf=".txt") + self.do_create(dir=".") + + def test_choose_directory(self): + # mkstemp can create directories in a user-selected directory + dir = tempfile.mkdtemp() + try: + self.do_create(dir=dir) + finally: + os.rmdir(dir) + +test_classes.append(test_mkstemp) + + +class test_mkdtemp(TC): + """Test mkdtemp().""" + + def do_create(self, dir=None, pre="", suf=""): + if dir is None: + dir = tempfile.gettempdir() + try: + name = tempfile.mkdtemp(dir=dir, prefix=pre, suffix=suf) + except: + self.failOnException("mkdtemp") + + try: + self.nameCheck(name, dir, pre, suf) + return name + except: + os.rmdir(name) + raise + + def test_basic(self): + # mkdtemp can create directories + os.rmdir(self.do_create()) + os.rmdir(self.do_create(pre="a")) + os.rmdir(self.do_create(suf="b")) + os.rmdir(self.do_create(pre="a", suf="b")) + os.rmdir(self.do_create(pre="aa", suf=".txt")) + + def test_basic_many(self): + # mkdtemp can create many directories (stochastic) + extant = range(TEST_FILES) + try: + for i in extant: + extant[i] = self.do_create(pre="aa") + finally: + for i in extant: + if(isinstance(i, basestring)): + os.rmdir(i) + + def test_choose_directory(self): + # mkdtemp can create directories in a user-selected directory + dir = tempfile.mkdtemp() + try: + os.rmdir(self.do_create(dir=dir)) + finally: + os.rmdir(dir) + + def test_mode(self): + # mkdtemp creates directories with the proper mode + if not has_stat: + return # ugh, can't use TestSkipped. + if os.name == 'java': + # Java doesn't support stating files for permissions + return + + dir = self.do_create() + try: + mode = stat.S_IMODE(os.stat(dir).st_mode) + mode &= 0777 # Mask off sticky bits inherited from /tmp + expected = 0700 + if sys.platform in ('win32', 'os2emx', 'mac'): + # There's no distinction among 'user', 'group' and 'world'; + # replicate the 'user' bits. + user = expected >> 6 + expected = user * (1 + 8 + 64) + self.assertEqual(mode, expected) + finally: + os.rmdir(dir) + +test_classes.append(test_mkdtemp) + + +class test_mktemp(TC): + """Test mktemp().""" + + # For safety, all use of mktemp must occur in a private directory. + # We must also suppress the RuntimeWarning it generates. + def setUp(self): + self.dir = tempfile.mkdtemp() + + def tearDown(self): + if self.dir: + os.rmdir(self.dir) + self.dir = None + + class mktemped: + _bflags = tempfile._bin_openflags + + def __init__(self, dir, pre, suf): + # XXX: Assign _unlink here, instead of as a class var. submit back + # to CPython + self._unlink = os.unlink + + self.name = tempfile.mktemp(dir=dir, prefix=pre, suffix=suf) + # Create the file. This will raise an exception if it's + # mysteriously appeared in the meanwhile. + fp = open(self.name, 'w+b') + fp.close() + # XXX: test_mktemp.tearDown expects the file to have been deleted + # (via __del__) by the time it's called, which is CPython specific + # garbage collection behavior. We need to delete it now in Jython + self._unlink(self.name) + + #def __del__(self): + # self._unlink(self.name) + + def do_create(self, pre="", suf=""): + try: + file = self.mktemped(self.dir, pre, suf) + except: + self.failOnException("mktemp") + + self.nameCheck(file.name, self.dir, pre, suf) + return file + + def test_basic(self): + # mktemp can choose usable file names + self.do_create() + self.do_create(pre="a") + self.do_create(suf="b") + self.do_create(pre="a", suf="b") + self.do_create(pre="aa", suf=".txt") + + def test_many(self): + # mktemp can choose many usable file names (stochastic) + extant = range(TEST_FILES) + for i in extant: + extant[i] = self.do_create(pre="aa") + +## def test_warning(self): +## # mktemp issues a warning when used +## warnings.filterwarnings("error", +## category=RuntimeWarning, +## message="mktemp") +## self.assertRaises(RuntimeWarning, +## tempfile.mktemp, dir=self.dir) + +test_classes.append(test_mktemp) + + +# We test _TemporaryFileWrapper by testing NamedTemporaryFile. + + +class test_NamedTemporaryFile(TC): + """Test NamedTemporaryFile().""" + + def do_create(self, dir=None, pre="", suf=""): + if dir is None: + dir = tempfile.gettempdir() + try: + file = tempfile.NamedTemporaryFile(dir=dir, prefix=pre, suffix=suf) + except: + self.failOnException("NamedTemporaryFile") + + self.nameCheck(file.name, dir, pre, suf) + return file + + + def test_basic(self): + # NamedTemporaryFile can create files + self.do_create() + self.do_create(pre="a") + self.do_create(suf="b") + self.do_create(pre="a", suf="b") + self.do_create(pre="aa", suf=".txt") + + def test_creates_named(self): + # NamedTemporaryFile creates files with names + f = tempfile.NamedTemporaryFile() + self.failUnless(os.path.exists(f.name), + "NamedTemporaryFile %s does not exist" % f.name) + + def test_del_on_close(self): + # A NamedTemporaryFile is deleted when closed + dir = tempfile.mkdtemp() + try: + f = tempfile.NamedTemporaryFile(dir=dir) + f.write('blat') + f.close() + self.failIf(os.path.exists(f.name), + "NamedTemporaryFile %s exists after close" % f.name) + finally: + os.rmdir(dir) + + def test_multiple_close(self): + # A NamedTemporaryFile can be closed many times without error + + f = tempfile.NamedTemporaryFile() + f.write('abc\n') + f.close() + try: + f.close() + f.close() + except: + self.failOnException("close") + + # How to test the mode and bufsize parameters? + +test_classes.append(test_NamedTemporaryFile) + + +class test_TemporaryFile(TC): + """Test TemporaryFile().""" + + def test_basic(self): + # TemporaryFile can create files + # No point in testing the name params - the file has no name. + try: + tempfile.TemporaryFile() + except: + self.failOnException("TemporaryFile") + + def test_has_no_name(self): + # TemporaryFile creates files with no names (on this system) + dir = tempfile.mkdtemp() + f = tempfile.TemporaryFile(dir=dir) + f.write('blat') + + # Sneaky: because this file has no name, it should not prevent + # us from removing the directory it was created in. + try: + os.rmdir(dir) + except: + ei = sys.exc_info() + # cleanup + f.close() + os.rmdir(dir) + self.failOnException("rmdir", ei) + + def test_multiple_close(self): + # A TemporaryFile can be closed many times without error + f = tempfile.TemporaryFile() + f.write('abc\n') + f.close() + try: + f.close() + f.close() + except: + self.failOnException("close") + + # How to test the mode and bufsize parameters? + + +if tempfile.NamedTemporaryFile is not tempfile.TemporaryFile: + test_classes.append(test_TemporaryFile) + +def test_main(): + test_support.run_unittest(*test_classes) + +if __name__ == "__main__": + test_main() Index: Lib/tempfile.py =================================================================== --- Lib/tempfile.py (revision 3454) +++ Lib/tempfile.py (working copy) @@ -1,189 +1,420 @@ -# XXX added to fix jython specific problem. Should be removed when real -# problem is fixed. -import java.io.File -"""Temporary files and filenames.""" +"""Temporary files. -# XXX This tries to be not UNIX specific, but I don't know beans about -# how to choose a temp directory or filename on MS-DOS or other -# systems so it may have to be changed... +This module provides generic, low- and high-level interfaces for +creating temporary files and directories. The interfaces listed +as "safe" just below can be used without fear of race conditions. +Those listed as "unsafe" cannot, and are provided for backward +compatibility only. -import os +This module also provides some data items to the user: -__all__ = ["mktemp", "TemporaryFile", "tempdir", "gettempprefix"] + TMP_MAX - maximum number of names that will be tried before + giving up. + template - the default prefix for all temporary names. + You may change this to control the default prefix. + tempdir - If this is set to a string before the first use of + any routine from this module, it will be considered as + another candidate location to store temporary files. +""" -# Parameters that the caller may set to override the defaults +__all__ = [ + "NamedTemporaryFile", "TemporaryFile", # high level safe interfaces + "mkstemp", "mkdtemp", # low level safe interfaces + "mktemp", # deprecated unsafe interface + "TMP_MAX", "gettempprefix", # constants + "tempdir", "gettempdir" + ] + + +# Imports. + +import os as _os +import errno as _errno +from random import Random as _Random + +if _os.name == 'mac': + import Carbon.Folder as _Folder + import Carbon.Folders as _Folders + +try: + import fcntl as _fcntl +except ImportError: + def _set_cloexec(fd): + pass +else: + def _set_cloexec(fd): + try: + flags = _fcntl.fcntl(fd, _fcntl.F_GETFD, 0) + except IOError: + pass + else: + # flags read successfully, modify + flags |= _fcntl.FD_CLOEXEC + _fcntl.fcntl(fd, _fcntl.F_SETFD, flags) + + +try: + import thread as _thread +except ImportError: + import dummy_thread as _thread +_allocate_lock = _thread.allocate_lock + +if _os.name == 'java': + _text_openflags = None + _bin_openflags = None +else: + _text_openflags = _os.O_RDWR | _os.O_CREAT | _os.O_EXCL + if hasattr(_os, 'O_NOINHERIT'): + _text_openflags |= _os.O_NOINHERIT + if hasattr(_os, 'O_NOFOLLOW'): + _text_openflags |= _os.O_NOFOLLOW + + _bin_openflags = _text_openflags + if hasattr(_os, 'O_BINARY'): + _bin_openflags |= _os.O_BINARY + +if hasattr(_os, 'TMP_MAX'): + TMP_MAX = _os.TMP_MAX +else: + TMP_MAX = 10000 + +template = "tmp" + tempdir = None -template = None -def gettempdir(): - """Function to calculate the directory to use.""" - global tempdir - if tempdir is not None: - return tempdir +# Internal routines. - # _gettempdir_inner deduces whether a candidate temp dir is usable by - # trying to create a file in it, and write to it. If that succeeds, - # great, it closes the file and unlinks it. There's a race, though: - # the *name* of the test file it tries is the same across all threads - # under most OSes (Linux is an exception), and letting multiple threads - # all try to open, write to, close, and unlink a single file can cause - # a variety of bogus errors (e.g., you cannot unlink a file under - # Windows if anyone has it open, and two threads cannot create the - # same file in O_EXCL mode under Unix). The simplest cure is to serialize - # calls to _gettempdir_inner. This isn't a real expense, because the - # first thread to succeed sets the global tempdir, and all subsequent - # calls to gettempdir() reuse that without trying _gettempdir_inner. - _tempdir_lock.acquire() +_once_lock = _allocate_lock() + +if hasattr(_os, "lstat"): + _stat = _os.lstat +elif hasattr(_os, "stat"): + _stat = _os.stat +else: + # Fallback. All we need is something that raises os.error if the + # file doesn't exist. + def _stat(fn): + try: + f = open(fn) + except IOError: + raise _os.error + f.close() + +def _exists(fn): try: - return _gettempdir_inner() - finally: - _tempdir_lock.release() + _stat(fn) + except _os.error: + return False + else: + return True -def _gettempdir_inner(): - """Function to calculate the directory to use.""" - global tempdir - if tempdir is not None: - return tempdir - try: - pwd = os.getcwd() - except (AttributeError, os.error): - pwd = os.curdir - attempdirs = ['/tmp', '/var/tmp', '/usr/tmp', pwd] - if os.name == 'nt': - attempdirs.insert(0, 'C:\\TEMP') - attempdirs.insert(0, '\\TEMP') - elif os.name == 'mac': - import macfs, MACFS +class _RandomNameSequence: + """An instance of _RandomNameSequence generates an endless + sequence of unpredictable strings which can safely be incorporated + into file names. Each string is six characters long. Multiple + threads can safely use the same instance at the same time. + + _RandomNameSequence is an iterator.""" + + characters = ("abcdefghijklmnopqrstuvwxyz" + + "ABCDEFGHIJKLMNOPQRSTUVWXYZ" + + "0123456789-_") + + def __init__(self): + self.mutex = _allocate_lock() + self.rng = _Random() + self.normcase = _os.path.normcase + + def __iter__(self): + return self + + def next(self): + m = self.mutex + c = self.characters + choose = self.rng.choice + + m.acquire() try: - refnum, dirid = macfs.FindFolder(MACFS.kOnSystemDisk, - MACFS.kTemporaryFolderType, 1) - dirname = macfs.FSSpec((refnum, dirid, '')).as_pathname() - attempdirs.insert(0, dirname) - except macfs.error: - pass - elif os.name == 'riscos': - scrapdir = os.getenv('Wimp$ScrapDir') - if scrapdir: - attempdirs.insert(0, scrapdir) + letters = [choose(c) for dummy in "123456"] + finally: + m.release() + + return self.normcase(''.join(letters)) + +def _candidate_tempdir_list(): + """Generate a list of candidate temporary directories which + _get_default_tempdir will try.""" + + dirlist = [] + + # First, try the environment. for envname in 'TMPDIR', 'TEMP', 'TMP': - if os.environ.has_key(envname): - attempdirs.insert(0, os.environ[envname]) - testfile = gettempprefix() + 'test' - for dir in attempdirs: + dirname = _os.getenv(envname) + if dirname: dirlist.append(dirname) + + # Failing that, try OS-specific locations. + if _os.name == 'mac': try: - filename = os.path.join(dir, testfile) - if os.name == 'posix': + fsr = _Folder.FSFindFolder(_Folders.kOnSystemDisk, + _Folders.kTemporaryFolderType, 1) + dirname = fsr.as_pathname() + dirlist.append(dirname) + except _Folder.error: + pass + elif _os.name == 'riscos': + dirname = _os.getenv('Wimp$ScrapDir') + if dirname: dirlist.append(dirname) + elif _os.name == 'nt': + dirlist.extend([ r'c:\temp', r'c:\tmp', r'\temp', r'\tmp' ]) + else: + dirlist.extend([ '/tmp', '/var/tmp', '/usr/tmp' ]) + + # As a last resort, the current directory. + try: + dirlist.append(_os.getcwd()) + except (AttributeError, _os.error): + dirlist.append(_os.curdir) + + return dirlist + +def _get_default_tempdir(): + """Calculate the default directory to use for temporary files. + This routine should be called exactly once. + + We determine whether or not a candidate temp dir is usable by + trying to create and write to a file in that directory. If this + is successful, the test file is deleted. To prevent denial of + service, the name of the test file must be randomized.""" + + namer = _RandomNameSequence() + dirlist = _candidate_tempdir_list() + flags = _text_openflags + + for dir in dirlist: + if dir != _os.curdir: + dir = _os.path.normcase(_os.path.abspath(dir)) + # Try only a few names per directory. + for seq in xrange(100): + name = namer.next() + filename = _os.path.join(dir, name) + if _os.name == 'java': + # XXX: Java always opens Files in _bin_openflags, and lacks low + # level access to file descriptors + fp = open(filename, 'w') + fp.write('blat') + fp.close() + _os.unlink(filename) + del fp + return dir + else: try: - fd = os.open(filename, - os.O_RDWR | os.O_CREAT | os.O_EXCL, 0700) - except OSError: - pass - else: - fp = os.fdopen(fd, 'w') + fd = _os.open(filename, flags, 0600) + fp = _os.fdopen(fd, 'w') fp.write('blat') fp.close() - os.unlink(filename) + _os.unlink(filename) del fp, fd - tempdir = dir - break - else: - fp = open(filename, 'w') - fp.write('blat') - fp.close() - os.unlink(filename) - tempdir = dir - break - except IOError: - pass + return dir + except (OSError, IOError), e: + if e[0] != _errno.EEXIST: + break # no point trying more names in this directory + pass + raise IOError, (_errno.ENOENT, + ("No usable temporary directory found in %s" % dirlist)) + +_name_sequence = None + +def _get_candidate_names(): + """Common setup sequence for all user-callable interfaces.""" + + global _name_sequence + if _name_sequence is None: + _once_lock.acquire() + try: + if _name_sequence is None: + _name_sequence = _RandomNameSequence() + finally: + _once_lock.release() + return _name_sequence + + +def _mkstemp_inner(dir, pre, suf, flags): + """Code common to mkstemp, TemporaryFile, and NamedTemporaryFile.""" + + names = _get_candidate_names() + + if _os.name == 'java': + # XXX: java.io.File.createTempFile requires minimum 3 character prefix + if len(pre) < 3: + namer = _RandomNameSequence() + pre += namer.next()[:3 - len(pre)] + # XXX: Let java create a secure file for us; Java lacks lower level + # access to file descriptors and creating a file with _os.O_CREAT | + # _os.O_EXCL. All we can return is the filename + import java.io.File + java_fp = java.io.File.createTempFile(pre, suf, + java.io.File(dir)) + fp = open(java_fp.path) + name = _os.path.abspath(fp.name) + fp.close() + return (None, name) + else: + for seq in xrange(TMP_MAX): + name = names.next() + file = _os.path.join(dir, pre + name + suf) + try: + fd = _os.open(file, flags, 0600) + _set_cloexec(fd) + return (fd, _os.path.abspath(file)) + except OSError, e: + if e.errno == _errno.EEXIST: + continue # try again + raise + + raise IOError, (_errno.EEXIST, "No usable temporary file name found") + + +# User visible interfaces. + +def gettempprefix(): + """Accessor for tempdir.template.""" + return template + +tempdir = None + +def gettempdir(): + """Accessor for tempdir.tempdir.""" + global tempdir if tempdir is None: - msg = "Can't find a usable temporary directory amongst " + `attempdirs` - raise IOError, msg + _once_lock.acquire() + try: + if tempdir is None: + tempdir = _get_default_tempdir() + finally: + _once_lock.release() return tempdir +def mkstemp(suffix="", prefix=template, dir=None, text=False): + """mkstemp([suffix, [prefix, [dir, [text]]]]) + User-callable function to create and return a unique temporary + file. The return value is a pair (fd, name) where fd is the + file descriptor returned by os.open, and name is the filename. -# template caches the result of gettempprefix, for speed, when possible. -# XXX unclear why this isn't "_template"; left it "template" for backward -# compatibility. -if os.name == "posix": - # We don't try to cache the template on posix: the pid may change on us - # between calls due to a fork, and on Linux the pid changes even for - # another thread in the same process. Since any attempt to keep the - # cache in synch would have to call os.getpid() anyway in order to make - # sure the pid hasn't changed between calls, a cache wouldn't save any - # time. In addition, a cache is difficult to keep correct with the pid - # changing willy-nilly, and earlier attempts proved buggy (races). - template = None + If 'suffix' is specified, the file name will end with that suffix, + otherwise there will be no suffix. -# Else the pid never changes, so gettempprefix always returns the same -# string. -elif os.name == "nt": - template = '~' + `os.getpid()` + '-' -elif os.name in ('mac', 'riscos'): - template = 'Python-Tmp-' -else: - template = 'tmp' # XXX might choose a better one + If 'prefix' is specified, the file name will begin with that prefix, + otherwise a default prefix is used. -def gettempprefix(): - """Function to calculate a prefix of the filename to use. + If 'dir' is specified, the file will be created in that directory, + otherwise a default directory is used. - This incorporates the current process id on systems that support such a - notion, so that concurrent processes don't generate the same prefix. + If 'text' is specified and true, the file is opened in text + mode. Else (the default) the file is opened in binary mode. On + some operating systems, this makes no difference. + + The file is readable and writable only by the creating user ID. + If the operating system uses permission bits to indicate whether a + file is executable, the file is executable by no one. The file + descriptor is not inherited by children of this process. + + Caller is responsible for deleting the file when done with it. """ - global template - if template is None: - return '@' + `os.getpid()` + '.' + if dir is None: + dir = gettempdir() + + if text: + flags = _text_openflags else: - return template + flags = _bin_openflags + return _mkstemp_inner(dir, prefix, suffix, flags) -def mktemp(suffix=""): - """User-callable function to return a unique temporary file name.""" - dir = gettempdir() - pre = gettempprefix() - while 1: - i = _counter.get_next() - file = os.path.join(dir, pre + str(i) + suffix) - if not os.path.exists(file): + +def mkdtemp(suffix="", prefix=template, dir=None): + """mkdtemp([suffix, [prefix, [dir]]]) + User-callable function to create and return a unique temporary + directory. The return value is the pathname of the directory. + + Arguments are as for mkstemp, except that the 'text' argument is + not accepted. + + The directory is readable, writable, and searchable only by the + creating user. + + Caller is responsible for deleting the directory when done with it. + """ + + if dir is None: + dir = gettempdir() + + names = _get_candidate_names() + + for seq in xrange(TMP_MAX): + name = names.next() + file = _os.path.join(dir, prefix + name + suffix) + try: + _os.mkdir(file, 0700) return file + except OSError, e: + if e.errno == _errno.EEXIST: + continue # try again + raise + raise IOError, (_errno.EEXIST, "No usable temporary directory name found") -class TemporaryFileWrapper: - """Temporary file wrapper +def mktemp(suffix="", prefix=template, dir=None): + """mktemp([suffix, [prefix, [dir]]]) + User-callable function to return a unique temporary file name. The + file is not created. - This class provides a wrapper around files opened for temporary use. - In particular, it seeks to automatically remove the file when it is - no longer needed. + Arguments are as for mkstemp, except that the 'text' argument is + not accepted. + + This function is unsafe and should not be used. The file name + refers to a file that did not exist at some point, but by the time + you get around to creating it, someone else may have beaten you to + the punch. """ - # Cache the unlinker so we don't get spurious errors at shutdown - # when the module-level "os" is None'd out. Note that this must - # be referenced as self.unlink, because the name TemporaryFileWrapper - # may also get None'd out before __del__ is called. +## from warnings import warn as _warn +## _warn("mktemp is a potential security risk to your program", +## RuntimeWarning, stacklevel=2) - # XXX: unlink = os.unlink does not work in jython, really that should be fixed and - # the original python class could be used. - if os.name == "java": - def unlink(self, path): - if not java.io.File(path).delete(): - raise OSError(0, "couldn't delete file", path) - else: - unlink = os.unlink + if dir is None: + dir = gettempdir() - def __init__(self, file, path): + names = _get_candidate_names() + for seq in xrange(TMP_MAX): + name = names.next() + file = _os.path.join(dir, prefix + name + suffix) + if not _exists(file): + return file + + raise IOError, (_errno.EEXIST, "No usable temporary filename found") + +class _TemporaryFileWrapper: + """Temporary file wrapper + + This class provides a wrapper around files opened for + temporary use. In particular, it seeks to automatically + remove the file when it is no longer needed. + """ + + def __init__(self, file, name): self.file = file - self.path = path - self.close_called = 0 + self.name = name + self.close_called = False - def close(self): - if not self.close_called: - self.close_called = 1 - self.file.close() - self.unlink(self.path) + # XXX: submit back to CPython + # Cache the unlinker so we don't get spurious errors at + # shutdown when the module-level "os" is None'd out. Note + # that this must be referenced as self.unlink, because the + # name TemporaryFileWrapper may also get None'd out before + # __del__ is called. + self.unlink = _os.unlink - def __del__(self): - self.close() - def __getattr__(self, name): file = self.__dict__['file'] a = getattr(file, name) @@ -191,65 +422,85 @@ setattr(self, name, a) return a + # NT provides delete-on-close as a primitive, so we don't need + # the wrapper to do anything special. We still use it so that + # file.name is useful (i.e. not "(fdopen)") with NamedTemporaryFile. + if _os.name != 'nt': -def TemporaryFile(mode='w+b', bufsize=-1, suffix=""): - """Create and return a temporary file (opened read-write by default).""" - name = mktemp(suffix) - if os.name == 'posix': - # Unix -- be very careful - fd = os.open(name, os.O_RDWR|os.O_CREAT|os.O_EXCL, 0700) - try: - os.unlink(name) - return os.fdopen(fd, mode, bufsize) - except: - os.close(fd) - raise - else: - # Non-unix -- can't unlink file that's still open, use wrapper - file = open(name, mode, bufsize) - return TemporaryFileWrapper(file, name) + def close(self): + if not self.close_called: + self.close_called = True + self.file.close() + self.unlink(self.name) -# In order to generate unique names, mktemp() uses _counter.get_next(). -# This returns a unique integer on each call, in a threadsafe way (i.e., -# multiple threads will never see the same integer). The integer will -# usually be a Python int, but if _counter.get_next() is called often -# enough, it will become a Python long. -# Note that the only names that survive this next block of code -# are "_counter" and "_tempdir_lock". + def __del__(self): + self.close() -class _ThreadSafeCounter: - def __init__(self, mutex, initialvalue=0): - self.mutex = mutex - self.i = initialvalue +def NamedTemporaryFile(mode='w+b', bufsize=-1, suffix="", + prefix=template, dir=None): + """Create and return a temporary file. + Arguments: + 'prefix', 'suffix', 'dir' -- as for mkstemp. + 'mode' -- the mode argument to os.fdopen (default "w+b"). + 'bufsize' -- the buffer size argument to os.fdopen (default -1). + The file is created as mkstemp() would do it. - def get_next(self): - self.mutex.acquire() - result = self.i - try: - newi = result + 1 - except OverflowError: - newi = long(result) + 1 - self.i = newi - self.mutex.release() - return result + Returns an object with a file-like interface; the name of the file + is accessible as file.name. The file will be automatically deleted + when it is closed. + """ -try: - import thread + if dir is None: + dir = gettempdir() -except ImportError: - class _DummyMutex: - def acquire(self): - pass + if 'b' in mode: + flags = _bin_openflags + else: + flags = _text_openflags - release = acquire + # Setting O_TEMPORARY in the flags causes the OS to delete + # the file when it is closed. This is only supported by Windows. + if _os.name == 'nt': + flags |= _os.O_TEMPORARY - _counter = _ThreadSafeCounter(_DummyMutex()) - _tempdir_lock = _DummyMutex() - del _DummyMutex + (fd, name) = _mkstemp_inner(dir, prefix, suffix, flags) + if _os.name == 'java': + file = open(name, mode, bufsize) + else: + file = _os.fdopen(fd, mode, bufsize) + return _TemporaryFileWrapper(file, name) +if _os.name != 'posix' or _os.sys.platform == 'cygwin': + # On non-POSIX and Cygwin systems, assume that we cannot unlink a file + # while it is open. + TemporaryFile = NamedTemporaryFile + else: - _counter = _ThreadSafeCounter(thread.allocate_lock()) - _tempdir_lock = thread.allocate_lock() - del thread + def TemporaryFile(mode='w+b', bufsize=-1, suffix="", + prefix=template, dir=None): + """Create and return a temporary file. + Arguments: + 'prefix', 'suffix', 'dir' -- as for mkstemp. + 'mode' -- the mode argument to os.fdopen (default "w+b"). + 'bufsize' -- the buffer size argument to os.fdopen (default -1). + The file is created as mkstemp() would do it. -del _ThreadSafeCounter + Returns an object with a file-like interface. The file has no + name, and will cease to exist when it is closed. + """ + + if dir is None: + dir = gettempdir() + + if 'b' in mode: + flags = _bin_openflags + else: + flags = _text_openflags + + (fd, name) = _mkstemp_inner(dir, prefix, suffix, flags) + try: + _os.unlink(name) + return _os.fdopen(fd, mode, bufsize) + except: + _os.close(fd) + raise