fix(util): wait for contended locks with relative filenames

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,

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()