| FazBrowse GitHub Viewer | Trending | | Home |
| Tools: [Download Repo ZIP] [Original HTTPS Page] |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -3,6 +3,7 @@ | |||
| 3 | 3 | ||
| 4 | 4 | """Index utilities.""" | |
| 5 | 5 | ||
| 6 | + import contextlib | ||
| 6 | 7 | from functools import wraps | |
| 7 | 8 | import os | |
| 8 | 9 | import os.path as osp | |
@@ -40,12 +41,10 @@ class TemporaryFileSwap: | |||
| 40 | 41 | ||
| 41 | 42 | def __init__(self, file_path: PathLike) -> None: | |
| 42 | 43 | self.file_path = file_path | |
| 43 | - self.tmp_file_path = str(self.file_path) + tempfile.mktemp("", "", "") | ||
| 44 | - # It may be that the source does not exist. | ||
| 45 | - try: | ||
| 46 | - os.rename(self.file_path, self.tmp_file_path) | ||
| 47 | - except OSError: | ||
| 48 | - pass | ||
| 44 | + fd, self.tmp_file_path = tempfile.mkstemp(prefix=self.file_path, dir="") | ||
| 45 | + os.close(fd) | ||
| 46 | + with contextlib.suppress(OSError): # It may be that the source does not exist. | ||
| 47 | + os.replace(self.file_path, self.tmp_file_path) | ||
| 49 | 48 | ||
| 50 | 49 | def __enter__(self) -> "TemporaryFileSwap": | |
| 51 | 50 | return self | |
@@ -57,10 +56,7 @@ def __exit__( | |||
| 57 | 56 | exc_tb: Optional[TracebackType], | |
| 58 | 57 | ) -> bool: | |
| 59 | 58 | if osp.isfile(self.tmp_file_path): | |
| 60 | - if os.name == "nt" and osp.exists(self.file_path): | ||
| 61 | - os.remove(self.file_path) | ||
| 62 | - os.rename(self.tmp_file_path, self.file_path) | ||
| 63 | - | ||
| 59 | + os.replace(self.tmp_file_path, self.file_path) | ||
| 64 | 60 | return False | |
| 65 | 61 | ||
| 66 | 62 | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -89,8 +89,7 @@ def with_rw_directory(func): | |||
| 89 | 89 | ||
| 90 | 90 | @wraps(func) | |
| 91 | 91 | def wrapper(self): | |
| 92 | - path = tempfile.mktemp(prefix=func.__name__) | ||
| 93 | - os.mkdir(path) | ||
| 92 | + path = tempfile.mkdtemp(prefix=func.__name__) | ||
| 94 | 93 | keep = False | |
| 95 | 94 | try: | |
| 96 | 95 | return func(self, path) | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -65,8 +65,7 @@ class TestBigRepoRW(TestBigRepoR): | |||
| 65 | 65 | def setUp(self): | |
| 66 | 66 | self.gitrwrepo = None | |
| 67 | 67 | super().setUp() | |
| 68 | - dirname = tempfile.mktemp() | ||
| 69 | - os.mkdir(dirname) | ||
| 68 | + dirname = tempfile.mkdtemp() | ||
| 70 | 69 | self.gitrwrepo = self.gitrorepo.clone(dirname, shared=True, bare=True, odbt=GitCmdObjectDB) | |
| 71 | 70 | self.puregitrwrepo = Repo(dirname, odbt=GitDB) | |
| 72 | 71 | ||
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -68,12 +68,11 @@ def test_base_object(self): | |||
| 68 | 68 | data = data_stream.read() | |
| 69 | 69 | assert data | |
| 70 | 70 | ||
| 71 | - tmpfilename = tempfile.mktemp(suffix="test-stream") | ||
| 72 | - with open(tmpfilename, "wb+") as tmpfile: | ||
| 71 | + with tempfile.NamedTemporaryFile(suffix="test-stream", delete=False) as tmpfile: | ||
| 73 | 72 | self.assertEqual(item, item.stream_data(tmpfile)) | |
| 74 | 73 | tmpfile.seek(0) | |
| 75 | 74 | self.assertEqual(tmpfile.read(), data) | |
| 76 | - os.remove(tmpfilename) | ||
| 75 | + os.remove(tmpfile.name) # Do it this way so we can inspect the file on failure. | ||
| 77 | 76 | # END for each object type to create | |
| 78 | 77 | ||
| 79 | 78 | # Each has a unique sha. | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -1,16 +1,14 @@ | |||
| 1 | 1 | # This module is part of GitPython and is released under the | |
| 2 | 2 | # 3-Clause BSD License: https://opensource.org/license/bsd-3-clause/ | |
| 3 | 3 | ||
| 4 | - import os | ||
| 4 | + import os.path as osp | ||
| 5 | 5 | import tempfile | |
| 6 | 6 | ||
| 7 | 7 | from git.objects import IndexObject | |
| 8 | 8 | from git.refs import RefLogEntry, RefLog | |
| 9 | 9 | from test.lib import TestBase, fixture_path | |
| 10 | 10 | from git.util import Actor, rmtree, hex_to_bin | |
| 11 | 11 | ||
| 12 | - import os.path as osp | ||
| 13 | - | ||
| 14 | 12 | ||
| 15 | 13 | class TestRefLog(TestBase): | |
| 16 | 14 | def test_reflogentry(self): | |
@@ -35,8 +33,7 @@ def test_reflogentry(self): | |||
| 35 | 33 | def test_base(self): | |
| 36 | 34 | rlp_head = fixture_path("reflog_HEAD") | |
| 37 | 35 | rlp_master = fixture_path("reflog_master") | |
| 38 | - tdir = tempfile.mktemp(suffix="test_reflogs") | ||
| 39 | - os.mkdir(tdir) | ||
| 36 | + tdir = tempfile.mkdtemp(suffix="test_reflogs") | ||
| 40 | 37 | ||
| 41 | 38 | rlp_master_ro = RefLog.path(self.rorepo.head) | |
| 42 | 39 | assert osp.isfile(rlp_master_ro) | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -667,11 +667,10 @@ def test_tag_to_full_tag_path(self): | |||
| 667 | 667 | self.assertEqual(value_errors, []) | |
| 668 | 668 | ||
| 669 | 669 | def test_archive(self): | |
| 670 | - tmpfile = tempfile.mktemp(suffix="archive-test") | ||
| 671 | - with open(tmpfile, "wb") as stream: | ||
| 670 | + with tempfile.NamedTemporaryFile("wb", suffix="archive-test", delete=False) as stream: | ||
| 672 | 671 | self.rorepo.archive(stream, "0.1.6", path="doc") | |
| 673 | 672 | assert stream.tell() | |
| 674 | - os.remove(tmpfile) | ||
| 673 | + os.remove(stream.name) # Do it this way so we can inspect the file on failure. | ||
| 675 | 674 | ||
| 676 | 675 | @mock.patch.object(Git, "_call_process") | |
| 677 | 676 | def test_should_display_blame_information(self, git): | |
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -359,48 +359,52 @@ def test_it_should_dashify(self): | |||
| 359 | 359 | self.assertEqual("foo", dashify("foo")) | |
| 360 | 360 | ||
| 361 | 361 | def test_lock_file(self): | |
| 362 | - my_file = tempfile.mktemp() | ||
| 363 | - lock_file = LockFile(my_file) | ||
| 364 | - assert not lock_file._has_lock() | ||
| 365 | - # Release lock we don't have - fine. | ||
| 366 | - lock_file._release_lock() | ||
| 362 | + with tempfile.TemporaryDirectory() as tdir: | ||
| 363 | + my_file = os.path.join(tdir, "my-lock-file") | ||
| 364 | + lock_file = LockFile(my_file) | ||
| 365 | + assert not lock_file._has_lock() | ||
| 366 | + # Release lock we don't have - fine. | ||
| 367 | + lock_file._release_lock() | ||
| 367 | 368 | ||
| 368 | - # Get lock. | ||
| 369 | - lock_file._obtain_lock_or_raise() | ||
| 370 | - assert lock_file._has_lock() | ||
| 369 | + # Get lock. | ||
| 370 | + lock_file._obtain_lock_or_raise() | ||
| 371 | + assert lock_file._has_lock() | ||
| 371 | 372 | ||
| 372 | - # Concurrent access. | ||
| 373 | - other_lock_file = LockFile(my_file) | ||
| 374 | - assert not other_lock_file._has_lock() | ||
| 375 | - self.assertRaises(IOError, other_lock_file._obtain_lock_or_raise) | ||
| 373 | + # Concurrent access. | ||
| 374 | + other_lock_file = LockFile(my_file) | ||
| 375 | + assert not other_lock_file._has_lock() | ||
| 376 | + self.assertRaises(IOError, other_lock_file._obtain_lock_or_raise) | ||
| 376 | 377 | ||
| 377 | - lock_file._release_lock() | ||
| 378 | - assert not lock_file._has_lock() | ||
| 378 | + lock_file._release_lock() | ||
| 379 | + assert not lock_file._has_lock() | ||
| 379 | 380 | ||
| 380 | - other_lock_file._obtain_lock_or_raise() | ||
| 381 | - self.assertRaises(IOError, lock_file._obtain_lock_or_raise) | ||
| 381 | + other_lock_file._obtain_lock_or_raise() | ||
| 382 | + self.assertRaises(IOError, lock_file._obtain_lock_or_raise) | ||
| 382 | 383 | ||
| 383 | - # Auto-release on destruction. | ||
| 384 | - del other_lock_file | ||
| 385 | - lock_file._obtain_lock_or_raise() | ||
| 386 | - lock_file._release_lock() | ||
| 384 | + # Auto-release on destruction. | ||
| 385 | + del other_lock_file | ||
| 386 | + lock_file._obtain_lock_or_raise() | ||
| 387 | + lock_file._release_lock() | ||
| 387 | 388 | ||
| 388 | 389 | def test_blocking_lock_file(self): | |
| 389 | - my_file = tempfile.mktemp() | ||
| 390 | - lock_file = BlockingLockFile(my_file) | ||
| 391 | - lock_file._obtain_lock() | ||
| 392 | - | ||
| 393 | - # Next one waits for the lock. | ||
| 394 | - start = time.time() | ||
| 395 | - wait_time = 0.1 | ||
| 396 | - wait_lock = BlockingLockFile(my_file, 0.05, wait_time) | ||
| 397 | - self.assertRaises(IOError, wait_lock._obtain_lock) | ||
| 398 | - elapsed = time.time() - start | ||
| 390 | + with tempfile.TemporaryDirectory() as tdir: | ||
| 391 | + my_file = os.path.join(tdir, "my-lock-file") | ||
| 392 | + lock_file = BlockingLockFile(my_file) | ||
| 393 | + lock_file._obtain_lock() | ||
| 394 | + | ||
| 395 | + # Next one waits for the lock. | ||
| 396 | + start = time.time() | ||
| 397 | + wait_time = 0.1 | ||
| 398 | + wait_lock = BlockingLockFile(my_file, 0.05, wait_time) | ||
| 399 | + self.assertRaises(IOError, wait_lock._obtain_lock) | ||
| 400 | + elapsed = time.time() - start | ||
| 401 | + | ||
| 399 | 402 | extra_time = 0.02 | |
| 400 | 403 | if os.name == "nt" or sys.platform == "cygwin": | |
| 401 | 404 | extra_time *= 6 # Without this, we get indeterministic failures on Windows. | |
| 402 | 405 | elif sys.platform == "darwin": | |
| 403 | 406 | extra_time *= 18 # The situation on macOS is similar, but with more delay. | |
| 407 | + | ||
| 404 | 408 | self.assertLess(elapsed, wait_time + extra_time) | |
| 405 | 409 | ||
| 406 | 410 | def test_user_id(self): | |
| Back | FazBrowse Home | New Git URL |
0 commit comments