# Imports.
-import atexit as _atexit
import functools as _functools
import warnings as _warnings
- import sys as _sys
import io as _io
import os as _os
+ import shutil as _shutil
import errno as _errno
from random import Random as _Random
-
-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 OSError:
- pass
- else:
- # flags read successfully, modify
- flags |= _fcntl.FD_CLOEXEC
- _fcntl.fcntl(fd, _fcntl.F_SETFD, flags)
-
++import weakref as _weakref
try:
import _thread
underlying file object, without adding a __del__ method to the
temporary file."""
- # Set here since __del__ checks it
- file = None
++ file = None # Set here since __del__ checks it
+ close_called = False
+
def __init__(self, file, name, delete=True):
self.file = file
self.name = name
in it are removed.
"""
+ # Handle mkdtemp raising an exception
+ name = None
++ _finalizer = None
+ _closed = False
+
def __init__(self, suffix="", prefix=template, dir=None):
- self._closed = False
- self.name = None # Handle mkdtemp raising an exception
self.name = mkdtemp(suffix, prefix, dir)
++ self._finalizer = _weakref.finalize(
++ self, self._cleanup, self.name,
++ warn_message="Implicitly cleaning up {!r}".format(self))
++
++ @classmethod
++ def _cleanup(cls, name, warn_message=None):
++ _shutil.rmtree(name)
++ if warn_message is not None:
++ _warnings.warn(warn_message, ResourceWarning)
++
def __repr__(self):
return "<{} {!r}>".format(self.__class__.__name__, self.name)
def __enter__(self):
return self.name
- def cleanup(self, _warn=False):
- def cleanup(self, _warn=False, _warnings=_warnings):
-- if self.name and not self._closed:
-- try:
- self._rmtree(self.name)
- _shutil.rmtree(self.name)
-- except (TypeError, AttributeError) as ex:
- # Issue #10188: Emit a warning on stderr
- # if the directory could not be cleaned
- # up due to missing globals
- if "None" not in str(ex):
- if "None" not in '%s' % (ex,):
-- raise
- print("ERROR: {!r} while cleaning up {!r}".format(ex, self,),
- file=_sys.stderr)
- return
- self._rmtree(self.name)
-- self._closed = True
- if _warn:
- self._warn("Implicitly cleaning up {!r}".format(self),
- ResourceWarning)
- if _warn and _warnings.warn:
- try:
- _warnings.warn("Implicitly cleaning up {!r}".format(self),
- ResourceWarning)
- except:
- if _is_running:
- raise
- # Don't raise an exception if modules needed for emitting
- # a warning are already cleaned in shutdown process.
--
def __exit__(self, exc, value, tb):
self.cleanup()
-- def __del__(self):
-- # Issue a ResourceWarning if implicit cleanup needed
-- self.cleanup(_warn=True)
-
- # XXX (ncoghlan): The following code attempts to make
- # this class tolerant of the module nulling out process
- # that happens during CPython interpreter shutdown
- # Alas, it doesn't actually manage it. See issue #10188
- _listdir = staticmethod(_os.listdir)
- _path_join = staticmethod(_os.path.join)
- _isdir = staticmethod(_os.path.isdir)
- _islink = staticmethod(_os.path.islink)
- _remove = staticmethod(_os.remove)
- _rmdir = staticmethod(_os.rmdir)
- _warn = _warnings.warn
--
- def _rmtree(self, path):
- def _rmtree(self, path, _OSError=OSError, _sep=_os.path.sep,
- _listdir=_os.listdir, _remove=_os.remove, _rmdir=_os.rmdir):
-- # Essentially a stripped down version of shutil.rmtree. We can't
-- # use globals because they may be None'ed out at shutdown.
- for name in self._listdir(path):
- fullname = self._path_join(path, name)
- try:
- isdir = self._isdir(fullname) and not self._islink(fullname)
- except OSError:
- isdir = False
- if isdir:
- self._rmtree(fullname)
- else:
- try:
- self._remove(fullname)
- except OSError:
- pass
- if not isinstance(path, str):
- _sep = _sep.encode()
-- try:
- self._rmdir(path)
- except OSError:
- for name in _listdir(path):
- fullname = path + _sep + name
- try:
- _remove(fullname)
- except _OSError:
- self._rmtree(fullname)
- _rmdir(path)
- except _OSError:
-- pass
-
-_is_running = True
-
-def _on_shutdown():
- global _is_running
- _is_running = False
++ def cleanup(self):
++ if self._finalizer is not None:
++ self._finalizer.detach()
++ if self.name is not None and not self._closed:
++ _shutil.rmtree(self.name)
++ self._closed = True
+
-_atexit.register(_on_shutdown)
finally:
os.rmdir(dir)
- @unittest.expectedFailure # See issue #10188
def test_del_on_shutdown(self):
# A TemporaryDirectory may be cleaned up during shutdown
- # Make sure it works with the relevant modules nulled out
with self.do_create() as dir:
- d = self.do_create(dir=dir)
- # Mimic the nulling out of modules that
- # occurs during system shutdown
- modules = [os, os.path]
- if has_stat:
- modules.append(stat)
- # Currently broken, so suppress the warning
- # that is otherwise emitted on stdout
- with support.captured_stderr() as err:
- with NulledModules(*modules):
- d.cleanup()
- # Currently broken, so stop spurious exception by
- # indicating the object has already been closed
- d._closed = True
- # And this assert will fail, as expected by the
- # unittest decorator...
- self.assertFalse(os.path.exists(d.name),
- "TemporaryDirectory %s exists after cleanup" % d.name)
- for mod in ('os', 'shutil', 'sys', 'tempfile', 'warnings'):
++ for mod in ('builtins', 'os', 'shutil', 'sys', 'tempfile', 'warnings'):
+ code = """if True:
++ import builtins
+ import os
+ import shutil
+ import sys
+ import tempfile
+ import warnings
+
+ tmp = tempfile.TemporaryDirectory(dir={dir!r})
+ sys.stdout.buffer.write(tmp.name.encode())
+
+ tmp2 = os.path.join(tmp.name, 'test_dir')
+ os.mkdir(tmp2)
+ with open(os.path.join(tmp2, "test.txt"), "w") as f:
+ f.write("Hello world!")
+
+ {mod}.tmp = tmp
+
+ warnings.filterwarnings("always", category=ResourceWarning)
+ """.format(dir=dir, mod=mod)
+ rc, out, err = script_helper.assert_python_ok("-c", code)
+ tmp_name = out.decode().strip()
+ self.assertFalse(os.path.exists(tmp_name),
+ "TemporaryDirectory %s exists after cleanup" % tmp_name)
+ err = err.decode('utf-8', 'backslashreplace')
+ self.assertNotIn("Exception ", err)
++ self.assertIn("ResourceWarning: Implicitly cleaning up", err)
def test_warnings_on_cleanup(self):
- # Two kinds of warning on shutdown
- # Issue 10888: may write to stderr if modules are nulled out
- # ResourceWarning will be triggered by __del__
+ # ResourceWarning will be triggered by __del__
with self.do_create() as dir:
- if os.sep != '\\':
- # Embed a backslash in order to make sure string escaping
- # in the displayed error message is dealt with correctly
- suffix = '\\check_backslash_handling'
- else:
- suffix = ''
- d = self.do_create(dir=dir, suf=suffix)
-
- #Check for the Issue 10888 message
- modules = [os, os.path]
- if has_stat:
- modules.append(stat)
- with support.captured_stderr() as err:
- with NulledModules(*modules):
- d.cleanup()
- message = err.getvalue().replace('\\\\', '\\')
- self.assertIn("while cleaning up", message)
- self.assertIn(d.name, message)
+ d = self.do_create(dir=dir, recurse=3)
+ name = d.name
# Check for the resource warning
with support.check_warnings(('Implicitly', ResourceWarning), quiet=False):
projects / python / commitdiff