Merge pull request #1770 from EliahKagan/less-mktemp

Replace some uses of the deprecated mktemp function

Author: Byron

git/index/util.py

@@ -3,6 +3,7 @@
 
 """Index utilities."""
 
+import contextlib
 from functools import wraps
 import os
 import os.path as osp
@@ -40,12 +41,10 @@ class TemporaryFileSwap:
 
     def __init__(self, file_path: PathLike) -> None:
         self.file_path = file_path
-        self.tmp_file_path = str(self.file_path) + tempfile.mktemp("", "", "")
-        # It may be that the source does not exist.
-        try:
-            os.rename(self.file_path, self.tmp_file_path)
-        except OSError:
-            pass
+        fd, self.tmp_file_path = tempfile.mkstemp(prefix=self.file_path, dir="")
+        os.close(fd)
+        with contextlib.suppress(OSError):  # It may be that the source does not exist.
+            os.replace(self.file_path, self.tmp_file_path)
 
     def __enter__(self) -> "TemporaryFileSwap":
         return self
@@ -57,10 +56,7 @@ def __exit__(
         exc_tb: Optional[TracebackType],
     ) -> bool:
         if osp.isfile(self.tmp_file_path):
-            if os.name == "nt" and osp.exists(self.file_path):
-                os.remove(self.file_path)
-            os.rename(self.tmp_file_path, self.file_path)
-
+            os.replace(self.tmp_file_path, self.file_path)
         return False
 
 

test/lib/helper.py

@@ -89,8 +89,7 @@ def with_rw_directory(func):
 
     @wraps(func)
     def wrapper(self):
-        path = tempfile.mktemp(prefix=func.__name__)
-        os.mkdir(path)
+        path = tempfile.mkdtemp(prefix=func.__name__)
         keep = False
         try:
             return func(self, path)

test/performance/lib.py

@@ -65,8 +65,7 @@ class TestBigRepoRW(TestBigRepoR):
     def setUp(self):
         self.gitrwrepo = None
         super().setUp()
-        dirname = tempfile.mktemp()
-        os.mkdir(dirname)
+        dirname = tempfile.mkdtemp()
         self.gitrwrepo = self.gitrorepo.clone(dirname, shared=True, bare=True, odbt=GitCmdObjectDB)
         self.puregitrwrepo = Repo(dirname, odbt=GitDB)
 

test/test_base.py

@@ -68,12 +68,11 @@ def test_base_object(self):
             data = data_stream.read()
             assert data
 
-            tmpfilename = tempfile.mktemp(suffix="test-stream")
-            with open(tmpfilename, "wb+") as tmpfile:
+            with tempfile.NamedTemporaryFile(suffix="test-stream", delete=False) as tmpfile:
                 self.assertEqual(item, item.stream_data(tmpfile))
                 tmpfile.seek(0)
                 self.assertEqual(tmpfile.read(), data)
-            os.remove(tmpfilename)
+            os.remove(tmpfile.name)  # Do it this way so we can inspect the file on failure.
         # END for each object type to create
 
         # Each has a unique sha.

test/test_reflog.py

@@ -1,16 +1,14 @@
 # This module is part of GitPython and is released under the
 # 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/
 
-import os
+import os.path as osp
 import tempfile
 
 from git.objects import IndexObject
 from git.refs import RefLogEntry, RefLog
 from test.lib import TestBase, fixture_path
 from git.util import Actor, rmtree, hex_to_bin
 
-import os.path as osp
-
 
 class TestRefLog(TestBase):
     def test_reflogentry(self):
@@ -35,8 +33,7 @@ def test_reflogentry(self):
     def test_base(self):
         rlp_head = fixture_path("reflog_HEAD")
         rlp_master = fixture_path("reflog_master")
-        tdir = tempfile.mktemp(suffix="test_reflogs")
-        os.mkdir(tdir)
+        tdir = tempfile.mkdtemp(suffix="test_reflogs")
 
         rlp_master_ro = RefLog.path(self.rorepo.head)
         assert osp.isfile(rlp_master_ro)

test/test_repo.py

@@ -667,11 +667,10 @@ def test_tag_to_full_tag_path(self):
         self.assertEqual(value_errors, [])
 
     def test_archive(self):
-        tmpfile = tempfile.mktemp(suffix="archive-test")
-        with open(tmpfile, "wb") as stream:
+        with tempfile.NamedTemporaryFile("wb", suffix="archive-test", delete=False) as stream:
             self.rorepo.archive(stream, "0.1.6", path="doc")
             assert stream.tell()
-        os.remove(tmpfile)
+        os.remove(stream.name)  # Do it this way so we can inspect the file on failure.
 
     @mock.patch.object(Git, "_call_process")
     def test_should_display_blame_information(self, git):

test/test_util.py

@@ -359,48 +359,52 @@ def test_it_should_dashify(self):
         self.assertEqual("foo", dashify("foo"))
 
     def test_lock_file(self):
-        my_file = tempfile.mktemp()
-        lock_file = LockFile(my_file)
-        assert not lock_file._has_lock()
-        # Release lock we don't have - fine.
-        lock_file._release_lock()
+        with tempfile.TemporaryDirectory() as tdir:
+            my_file = os.path.join(tdir, "my-lock-file")
+            lock_file = LockFile(my_file)
+            assert not lock_file._has_lock()
+            # Release lock we don't have - fine.
+            lock_file._release_lock()
 
-        # Get lock.
-        lock_file._obtain_lock_or_raise()
-        assert lock_file._has_lock()
+            # Get lock.
+            lock_file._obtain_lock_or_raise()
+            assert lock_file._has_lock()
 
-        # Concurrent access.
-        other_lock_file = LockFile(my_file)
-        assert not other_lock_file._has_lock()
-        self.assertRaises(IOError, other_lock_file._obtain_lock_or_raise)
+            # Concurrent access.
+            other_lock_file = LockFile(my_file)
+            assert not other_lock_file._has_lock()
+            self.assertRaises(IOError, other_lock_file._obtain_lock_or_raise)
 
-        lock_file._release_lock()
-        assert not lock_file._has_lock()
+            lock_file._release_lock()
+            assert not lock_file._has_lock()
 
-        other_lock_file._obtain_lock_or_raise()
-        self.assertRaises(IOError, lock_file._obtain_lock_or_raise)
+            other_lock_file._obtain_lock_or_raise()
+            self.assertRaises(IOError, lock_file._obtain_lock_or_raise)
 
-        # Auto-release on destruction.
-        del other_lock_file
-        lock_file._obtain_lock_or_raise()
-        lock_file._release_lock()
+            # Auto-release on destruction.
+            del other_lock_file
+            lock_file._obtain_lock_or_raise()
+            lock_file._release_lock()
 
     def test_blocking_lock_file(self):
-        my_file = tempfile.mktemp()
-        lock_file = BlockingLockFile(my_file)
-        lock_file._obtain_lock()
-
-        # Next one waits for the lock.
-        start = time.time()
-        wait_time = 0.1
-        wait_lock = BlockingLockFile(my_file, 0.05, wait_time)
-        self.assertRaises(IOError, wait_lock._obtain_lock)
-        elapsed = time.time() - start
+        with tempfile.TemporaryDirectory() as tdir:
+            my_file = os.path.join(tdir, "my-lock-file")
+            lock_file = BlockingLockFile(my_file)
+            lock_file._obtain_lock()
+
+            # Next one waits for the lock.
+            start = time.time()
+            wait_time = 0.1
+            wait_lock = BlockingLockFile(my_file, 0.05, wait_time)
+            self.assertRaises(IOError, wait_lock._obtain_lock)
+            elapsed = time.time() - start
+
         extra_time = 0.02
         if os.name == "nt" or sys.platform == "cygwin":
             extra_time *= 6  # Without this, we get indeterministic failures on Windows.
         elif sys.platform == "darwin":
             extra_time *= 18  # The situation on macOS is similar, but with more delay.
+
         self.assertLess(elapsed, wait_time + extra_time)
 
     def test_user_id(self):