PidFileLock: Add more tests (#105)

Coverage is 96.8%

Co-authored-by: Sushant Gaurav <[email protected]>

Author: tomschr

src/docbuild/utils/pidlock.py

@@ -122,23 +122,30 @@ def __exit__(self, exc_type, exc_value, traceback) -> None:
 
         # 1. Release the fcntl lock
         try:
-            # Must use the file descriptor (handle.fileno()) for fcntl.flock
+            # Must use the file descriptor (handle.fileno()) for fcntl.flockq
             fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
-        except Exception:
-            # Non-critical, but log if needed
-            pass
+        except OSError as e:
+            # Log OS-level failures releasing the lock; don't raise from __exit__
+            log.warning(
+                "Failed to release fcntl lock for %s: %s", self.resource_path, e
+            )
 
         # 2. Close the file handle
         try:
             handle.close()
-        except Exception:
-            pass
+        except (OSError, ValueError) as e:
+            # Only handle known close-related errors (OS-level errors and ValueError
+            # if the file object is in an invalid state). Allow other unexpected
+            # exceptions to propagate.
+            log.warning("Failed to close lock file handle %s: %s", self._lock_path, e)
 
         # 3. Remove the lock file
         try:
             self._lock_path.unlink(missing_ok=True)
             log.debug("Released fcntl lock and removed file %s", self._lock_path)
-        except Exception as e:
+        except OSError as e:
+            # Only catch OS-level errors from unlink (e.g. permission issues);
+            # allow other unexpected exceptions to propagate.
             log.warning(f"Failed to remove lock file {self._lock_path}: {e}")
 
         self._handle = None

tests/utils/test_pidlock.py

@@ -1,15 +1,19 @@
 """Tests for the PidFileLock utility."""
 
 import errno
-import os
-import pytest
 import logging
 import multiprocessing as mp
-import time
-import platform
-from pathlib import Path
-from docbuild.utils.pidlock import PidFileLock, LockAcquisitionError
 from multiprocessing import Event
+import os
+from pathlib import Path
+import platform
+import time
+
+import pytest
+from unittest.mock import patch, Mock
+import docbuild.utils.pidlock as pidlock_mod
+import builtins
+from docbuild.utils.pidlock import LockAcquisitionError, PidFileLock
 
 # Define a shared marker to skip the test if the OS is macOS (Darwin)
 skip_macos = pytest.mark.skipif(
@@ -35,7 +39,7 @@ def _mp_lock_holder(resource_path: Path, lock_dir: Path, lock_path: Path, done_e
         with lock:
             lock_path.touch()
             done_event.wait()
-            
+
     except Exception:
         pass
 
@@ -69,7 +73,7 @@ def test_context_manager(lock_setup):
     assert not lock.lock_path.exists()
 
 
-@skip_macos 
+@skip_macos
 def test_lock_prevents_concurrent_access_in_separate_process(lock_setup):
     """Test that two separate processes cannot acquire the same lock simultaneously."""
 
@@ -175,7 +179,7 @@ def test_acquire_critical_oserror(monkeypatch, tmp_path):
 
     # Mock the built-in open to fail with EACCES, simulating a permissions error
     def mocked_builtin_open(path, mode):
-        # We now rely on monkeypatching os.open directly in pidlock.py, 
+        # We now rely on monkeypatching os.open directly in pidlock.py,
         # but for this test, we mock builtins.open if os.open is not used directly.
         # Since pidlock.py now uses open(self._lock_path, 'w+'), mocking builtins.open is correct
         raise OSError(errno.EACCES, "Access denied")
@@ -189,4 +193,243 @@ def mocked_builtin_open(path, mode):
     # Check that the RuntimeError message contains the expected text
     error_message = str(exc_info.value)
     assert "Cannot acquire lock:" in error_message
-    assert "Access denied" in error_message
+    assert "Access denied" in error_message
+
+
+def test_pidfilelock_singleton_per_lock_path(tmp_path):
+    """Constructing PidFileLock twice for the same resource should return the same instance."""
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+
+    lock1 = PidFileLock(resource, lock_dir=lock_dir)
+    lock2 = PidFileLock(resource, lock_dir=lock_dir)
+
+    # same computed lock_path and same object returned
+    assert lock1 is lock2
+    assert lock1.lock_path == lock2.lock_path
+
+
+def test_flock_eagain_raises_lockacquisitionerror(tmp_path):
+    """If fcntl.flock raises EAGAIN/EWOULDBLOCK, a LockAcquisitionError is raised
+    and the module's cleanup path (closing the opened handle) is executed.
+    Use a real file handle (no patch of open) so the actual close() source line
+    in pidlock.py is executed and counted by coverage.
+    """
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+    lock_dir.mkdir()
+
+    def raise_eagain(fd, op):
+        raise OSError(errno.EAGAIN, "Resource temporarily unavailable")
+
+    # Only patch the flock call; let pidlock.open create a real file handle so
+    # the except branch that calls handle.close() runs in pidlock.py.
+    with patch.object(pidlock_mod.fcntl, "flock", raise_eagain):
+        with pytest.raises(LockAcquisitionError):
+            with PidFileLock(resource, lock_dir=lock_dir):
+                pass
+
+
+def test_open_eacces_raises_runtimeerror(tmp_path):
+    """If opening the lock file fails with EACCES/EPERM, a RuntimeError is raised."""
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+
+    def fake_open(*args, **kwargs):
+        raise OSError(errno.EACCES, "Permission denied")
+
+    with patch.object(pidlock_mod, "open", fake_open):
+        with pytest.raises(RuntimeError, match="Cannot acquire lock"):
+            with PidFileLock(resource, lock_dir=lock_dir):
+                pass
+
+
+def test_open_eacces_raises_runtimeerror_via_builtins(tmp_path):
+    """Ensure the EACCES/EPERM open failure branch raises RuntimeError (covers the raise RuntimeError line)."""
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+    lock_dir.mkdir()
+    resource.write_text("dummy")
+    lock = PidFileLock(resource, lock_dir)
+
+    def fake_open(*args, **kwargs):
+        raise OSError(errno.EACCES, "Permission denied (simulated)")
+
+    # Patch the actual builtins.open used by pidlock.open(...) to ensure the except branch is hit
+    with patch.object(builtins, "open", fake_open):
+        with pytest.raises(RuntimeError, match="Cannot acquire lock"):
+            with lock:
+                pass
+
+
+def test_open_other_oserror_reraises_original_exception(tmp_path):
+    """If open() raises an OSError not handled by special branches, it should be re-raised."""
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+
+    def fake_open(*args, **kwargs):
+        raise OSError(errno.EINVAL, "Invalid argument")
+
+    with patch.object(pidlock_mod, "open", fake_open):
+        with pytest.raises(OSError) as excinfo:
+            with PidFileLock(resource, lock_dir=lock_dir):
+                pass
+
+    # Ensure the original errno is preserved (covers the final `raise e` branch)
+    assert excinfo.value.errno == errno.EINVAL
+
+
+def test_enter_handles_flock_eagain_closes_handle(tmp_path):
+    """When flock raises EAGAIN after open succeeded, the opened handle is closed and LockAcquisitionError is raised."""
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+    # Create fake handle and ensure it has a close() we can assert
+    fake_handle = Mock()
+    fake_handle.fileno.return_value = 123
+    fake_handle.seek.return_value = None
+    fake_handle.truncate.return_value = None
+    fake_handle.write.return_value = None
+    fake_handle.flush.return_value = None
+    fake_handle.close.return_value = None
+
+    def raise_eagain(fd, op):
+        raise OSError(errno.EAGAIN, "Resource temporarily unavailable")
+
+    with patch.object(pidlock_mod, "open", lambda *a, **k: fake_handle):
+        with patch.object(pidlock_mod.fcntl, "flock", raise_eagain):
+            with pytest.raises(LockAcquisitionError):
+                with PidFileLock(resource, lock_dir=lock_dir):
+                    pass
+
+    fake_handle.close.assert_called_once()
+
+
+def test_enter_open_eacces_raises_runtimeerror(tmp_path):
+    """If module-level open raises EACCES, __enter__ raises a RuntimeError wrapping that OSError."""
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+
+    def fake_open(*args, **kwargs):
+        raise OSError(errno.EACCES, "Permission denied (simulated)")
+
+    with patch.object(pidlock_mod, "open", fake_open):
+        with pytest.raises(RuntimeError, match="Cannot acquire lock"):
+            with PidFileLock(resource, lock_dir=lock_dir):
+                pass
+
+
+def test_exit_flock_unlock_oserror_is_handled(tmp_path):
+    """If releasing the fcntl lock (second flock call) raises OSError, __exit__ should catch and log but not raise."""
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+    lock_dir.mkdir()
+    calls = {"n": 0}
+
+    def flock_first_ok_then_raise(fd, op):
+        calls["n"] += 1
+        if calls["n"] == 1:
+            return None  # __enter__ flock succeeds
+        # second call (unlock) raises
+        raise OSError(errno.EIO, "I/O error on unlock")
+
+    with patch.object(pidlock_mod.fcntl, "flock", flock_first_ok_then_raise):
+        # Should not raise on context exit even though unlock raises
+        with PidFileLock(resource, lock_dir=lock_dir):
+            assert Path(lock_dir).exists()
+        # after exit, lock file cleaned up (unlink missing_ok may remove it)
+        assert not PidFileLock(resource, lock_dir).lock_path.exists()
+
+
+def test_exit_handle_close_raises_is_handled(tmp_path):
+    """If handle.close() raises OSError, __exit__ should catch and log and continue cleanup."""
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+    lock_dir.mkdir()
+
+    # fake handle whose close() will raise OSError
+    class BadHandle:
+        def fileno(self):
+            return 1
+        def seek(self, *_):
+            return None
+        def truncate(self, *_):
+            return None
+        def write(self, *_):
+            return None
+        def flush(self):
+            return None
+        def close(self):
+            raise OSError(errno.EIO, "close failed")
+
+    bad_handle = BadHandle()
+
+    # Make flock a no-op so __enter__ succeeds
+    with patch.object(pidlock_mod.fcntl, "flock", lambda *a, **k: None):
+        with patch.object(pidlock_mod, "open", lambda *a, **k: bad_handle):
+            # Should not raise on context exit despite close() raising
+            with PidFileLock(resource, lock_dir=lock_dir):
+                pass
+
+
+def test_exit_unlink_raises_oserror_is_handled(tmp_path):
+    """If unlink() raises OSError during __exit__, it should be caught and logged without raising."""
+    resource = tmp_path / "resource.txt"
+    lock_dir = tmp_path / "locks"
+    lock_dir.mkdir()
+
+    # Use a normal file handle and normal flock to acquire lock
+    with patch.object(pidlock_mod.fcntl, "flock", lambda *a, **k: None):
+        lock = PidFileLock(resource, lock_dir=lock_dir)
+
+        # Patch the Path.unlink implementation used by pidlock so unlink() raises OSError
+        def bad_unlink(self, missing_ok=True):
+            raise OSError(errno.EIO, "unlink failed")
+
+        with patch.object(pidlock_mod.Path, "unlink", bad_unlink):
+            # enter the context; when __exit__ calls unlink(), it will raise and be handled
+            with lock:
+                pass
+
+    # exit should not raise even though unlink raised
+    assert not lock._lock_acquired
+    assert lock._handle is None
+
+
+def test_enter_open_eacces_on_fresh_instance(tmp_path):
+    """Force the EACCES/EPERM branch on a fresh PidFileLock instance to cover the RuntimeError path."""
+    resource = tmp_path / "resource_eacces.txt"
+    lock_dir = tmp_path / "locks_eacces"
+
+    def fake_open(*args, **kwargs):
+        raise OSError(errno.EACCES, "Permission denied (forced)")
+
+    # Patch the module-level open used by pidlock for this fresh instance
+    with patch.object(pidlock_mod, "open", fake_open):
+        with pytest.raises(RuntimeError, match="Cannot acquire lock"):
+            with PidFileLock(resource, lock_dir=lock_dir):
+                pass
+
+
+def test_enter_flock_eagain_no_handle(tmp_path):
+    """
+    Simulate EAGAIN/EWOULDBLOCK being raised when 'handle' is None to hit the
+    'if handle: handle.close()' branch where 'handle' is None.
+    """
+
+    resource = tmp_path / 'resource.txt'
+    lock_dir = tmp_path / 'locks'
+
+    # Mock open to return a handle
+    real_handle = open(tmp_path / 'lockfile.tmp', 'w+')
+
+    # Using two mocks: one for open to succeed, one for flock to fail.
+    # Patch 'open' in a way that allows controlling the assignment of 'handle'.
+
+    # Simulate open failing with EAGAIN directly, which also forces 'handle' to be None.
+    def fake_open_with_eagain(*args, **kwargs):
+        raise OSError(errno.EAGAIN, 'Simulated open error matching flock failure')
+
+    with patch.object(pidlock_mod, 'open', fake_open_with_eagain):
+        with pytest.raises(LockAcquisitionError, match='Resource is locked'):
+            with PidFileLock(resource, lock_dir=lock_dir):
+                pass