diff --git a/git/index/util.py b/git/index/util.py index b1aaa58fd..1c3b1c4ad 100644 --- a/git/index/util.py +++ b/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 diff --git a/test/lib/helper.py b/test/lib/helper.py index 58d96534a..d662b632d 100644 --- a/test/lib/helper.py +++ b/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) diff --git a/test/performance/lib.py b/test/performance/lib.py index ceee6c2a1..d08e1027f 100644 --- a/test/performance/lib.py +++ b/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) diff --git a/test/test_base.py b/test/test_base.py index cdf82b74d..74f342071 100644 --- a/test/test_base.py +++ b/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. diff --git a/test/test_reflog.py b/test/test_reflog.py index 625466d40..1bd2e5dab 100644 --- a/test/test_reflog.py +++ b/test/test_reflog.py @@ -1,7 +1,7 @@ # 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 @@ -9,8 +9,6 @@ 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) diff --git a/test/test_repo.py b/test/test_repo.py index 007b8ecb0..465bb2574 100644 --- a/test/test_repo.py +++ b/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): diff --git a/test/test_util.py b/test/test_util.py index 1cb255cfe..6616e1067 100644 --- a/test/test_util.py +++ b/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):