diff --git a/git/util.py b/git/util.py index 61ddab861..196e95d87 100644 --- a/git/util.py +++ b/git/util.py @@ -1310,7 +1310,7 @@ def _obtain_lock(self) -> None: # synity check: if the directory leading to the lockfile is not # readable anymore, raise an exception curtime = time.time() - if not osp.isdir(osp.dirname(self._lock_file_path())): + if not osp.isdir(osp.dirname(self._lock_file_path()) or osp.curdir): msg = "Directory containing the lockfile %r was not readable anymore after waiting %g seconds" % ( self._lock_file_path(), curtime - starttime, diff --git a/test/test_util.py b/test/test_util.py index ca8e006f3..33e1e54f1 100644 --- a/test/test_util.py +++ b/test/test_util.py @@ -862,3 +862,40 @@ def test_mode_str_to_int_accepts_bytes(): assert mode_str_to_int(b"100644") == 0o100644 assert mode_str_to_int("644") == 0o644 assert mode_str_to_int(b"120000") == 0o120000 + + +@pytest.mark.parametrize("path_type", [str, pathlib.Path]) +@pytest.mark.parametrize("relative", [True, False]) +def test_blocking_lock_waits_for_existing_holder(tmp_path, monkeypatch, path_type, relative): + monkeypatch.chdir(tmp_path) + path = path_type("resource" if relative else tmp_path / "resource") + owner = LockFile(path) + waiter = BlockingLockFile(path, check_interval_s=0.01, max_block_time_s=1) + owner._obtain_lock() + try: + with mock.patch.object(time, "sleep", side_effect=lambda _: owner._release_lock()) as sleep: + waiter._obtain_lock() + sleep.assert_called_once_with(0.01) + assert waiter._has_lock() + finally: + owner._release_lock() + waiter._release_lock() + + +@pytest.mark.parametrize("path_type", [str, pathlib.Path]) +def test_blocking_relative_lock_reports_timeout(tmp_path, monkeypatch, path_type): + monkeypatch.chdir(tmp_path) + owner = LockFile(path_type("resource")) + waiter = BlockingLockFile(path_type("resource"), max_block_time_s=0) + owner._obtain_lock() + try: + with pytest.raises(OSError, match="Waited .* seconds for lock"): + waiter._obtain_lock() + finally: + owner._release_lock() + + +def test_blocking_lock_still_rejects_missing_directory(tmp_path): + waiter = BlockingLockFile(tmp_path / "missing" / "resource") + with pytest.raises(OSError, match="Directory containing the lockfile"): + waiter._obtain_lock()